# week8-homework

**Repository Path**: zero-night-01/week8-homework

## Basic Information

- **Project Name**: week8-homework
- **Description**: No description available
- **Primary Language**: Scala
- **License**: Not specified
- **Default Branch**: master
- **Homepage**: None
- **GVP Project**: No

## Statistics

- **Stars**: 0
- **Forks**: 0
- **Created**: 2022-04-23
- **Last Updated**: 2022-05-08

## Categories & Tags

**Categories**: Uncategorized

**Tags**: None

## README

# week8-homework

#### 作业说明

作业一
> 为 Spark SQL 添加一条自定义命令<br>
> - SHOW VERSION;<br>
> - 显示当前 Spark 版本和 Java 版本。

说明：需要下载 Spark 源码，此处以如下版本为例：
- Spark 3.2.1
- Scala 2.12
- Java 8

首先需要将 `SHOW VERSION` 注册为 `statement`，将 `VERSION` 注册为关键字，并绑定用户输入。

操作：在下载的源码中，找到 `sql/catalyst/src/main/antlr4/org/apache/spark/sql/catalyst/parser/SqlBase.g4` 并新增如下内容
```antlrv4
statement
    : query   
    | SHOW VERSION     #showVersion
    
ansiNonReserved
    | VERSION

nonReserved
    | VERSION
    
//--SPARK-KEYWORD-LIST-START
VERSION : 'VERSION';
```

接着在`spark-3.2.1` 根目录下运行命令：
```shell
build/mvn antlr4:antlr4 -f pom.xml
```
生成需要用到的上下文类等内容

接着实现对应的 `visit` 方法，如下：

在 `sql/core/src/main/scala/org/apache/spark/sql/execution/SparkSqlParser.scala` 中新增如下代码

```scala
class SparkSqlAstBuilder extends AstBuilder {
  ...
  override def visitShowVersion(ctx: ShowVersionContext): LogicalPlan = withOrigin(ctx) {
    ShowVersionCommand()
  }
  ...
}
```

在任意能访问到的地方创建 `ShowVersionCommand`，如下：

```scala
case class ShowVersionCommand() extends LeafRunnableCommand {
  override val output: Seq[Attribute] =
    Seq(AttributeReference("version", StringType)())

  override def run(sparkSession: SparkSession): Seq[Row] = {
    val sparkVersion = sparkSession.version
    val javaVersion = System.getProperty("java.version")
    val outputString = "spark version: %s, java version: %s".format(sparkVersion, javaVersion)
    Seq(Row(outputString))
  }
}
```

完成后进行编译：
```shell
build/mvn clean package -DskipTests -Phive -Phive-thriftserver
```

成功后进入项目内的 `SparkSQL`：
```shell
bin/spark-sql
```

验证功能：
```shell
spark-sql> show version;
spark version: 3.2.1, java version: 1.8.0_332
Time taken: 1.404 seconds, Fetched 1 row(s)
spark-sql> 
```

作业二

准备工作
```sparksql
set spark.sql.planChangeLog.level=WARN;
```

题目一

```sparksql
CREATE TABLE sales (
    id int,
    customerId int,
    amountPaid double
);
EXPLAIN
SELECT a.customerId
FROM (
  SELECT customerId,
         amountPaid as amount
    FROM sales
   WHERE amountPaid > 100.0 AND 1 = '1'
) a
WHERE a.customerId < 20;
```

日志片段

```shell
=== Applying Rule org.apache.spark.sql.catalyst.optimizer.PushDownPredicates ===
 Project [customerId#618]                                                                                                                                                         Project [customerId#618]
!+- Filter (customerId#618 < 20)                                                                                                                                                  +- Project [customerId#618, amountPaid#619 AS amount#611]
!   +- Project [customerId#618, amountPaid#619 AS amount#611]                                                                                                                        +- Filter (((amountPaid#619 > cast(100.0 as double)) AND (1 = cast(1 as int))) AND (customerId#618 < 20))
!      +- Filter ((amountPaid#619 > cast(100.0 as double)) AND (1 = cast(1 as int)))                                                                                                    +- HiveTableRelation [`default`.`sales`, org.apache.hadoop.hive.serde2.lazy.LazySimpleSerDe, Data Cols: [id#617, customerId#618, amountPaid#619], Partition Cols: []]
!         +- HiveTableRelation [`default`.`sales`, org.apache.hadoop.hive.serde2.lazy.LazySimpleSerDe, Data Cols: [id#617, customerId#618, amountPaid#619], Partition Cols: []]   
           
22/05/02 17:16:58 WARN PlanChangeLogger: 
=== Applying Rule org.apache.spark.sql.catalyst.optimizer.CollapseProject ===
 Project [customerId#618]                                                                                                                                                      Project [customerId#618]
!+- Project [customerId#618]                                                                                                                                                   +- Filter (((amountPaid#619 > cast(100.0 as double)) AND (1 = cast(1 as int))) AND (customerId#618 < 20))
!   +- Filter (((amountPaid#619 > cast(100.0 as double)) AND (1 = cast(1 as int))) AND (customerId#618 < 20))                                                                     +- HiveTableRelation [`default`.`sales`, org.apache.hadoop.hive.serde2.lazy.LazySimpleSerDe, Data Cols: [id#617, customerId#618, amountPaid#619], Partition Cols: []]
!      +- HiveTableRelation [`default`.`sales`, org.apache.hadoop.hive.serde2.lazy.LazySimpleSerDe, Data Cols: [id#617, customerId#618, amountPaid#619], Partition Cols: []]   

22/05/02 17:16:58 WARN PlanChangeLogger: 
=== Applying Rule org.apache.spark.sql.catalyst.optimizer.BooleanSimplification ===
 Project [customerId#618]                                                                                                                                                   Project [customerId#618]
!+- Filter (((amountPaid#619 > 100.0) AND true) AND (customerId#618 < 20))                                                                                                  +- Filter ((amountPaid#619 > 100.0) AND (customerId#618 < 20))
    +- HiveTableRelation [`default`.`sales`, org.apache.hadoop.hive.serde2.lazy.LazySimpleSerDe, Data Cols: [id#617, customerId#618, amountPaid#619], Partition Cols: []]      +- HiveTableRelation [`default`.`sales`, org.apache.hadoop.hive.serde2.lazy.LazySimpleSerDe, Data Cols: [id#617, customerId#618, amountPaid#619], Partition Cols: []]

```

注：Spark 3.0+ 中 PushDownPredicates 实现使用了 CombineFilters

题目二

```sparksql
EXPLAIN
SELECT DISTINCT a.customerId
FROM (
  SELECT customerId,
         amountPaid as amount,
         1.0 x, 'abc' y, Now() z
    FROM sales
   WHERE amountPaid > 100.0 AND 1 = '1'
   ORDER BY x, y, 3
) a
WHERE a.customerId < 20
EXCEPT SELECT customerId FROM sales;

```

日志片段

```shell
=== Applying Rule org.apache.spark.sql.catalyst.optimizer.ReplaceExceptWithAntiJoin ===
!Except false                                                                                                                                                                              Distinct
!:- Distinct                                                                                                                                                                               +- Join LeftAnti, (customerId#663 <=> customerId#666)
!:  +- Project [customerId#663]                                                                                                                                                               :- Distinct
!:     +- Filter (customerId#663 < 20)                                                                                                                                                        :  +- Project [customerId#663]
!:        +- Sort [x#654 ASC NULLS FIRST, y#655 ASC NULLS FIRST, x#654 ASC NULLS FIRST], true                                                                                                 :     +- Filter (customerId#663 < 20)
!:           +- Project [customerId#663, amountPaid#664 AS amount#653, 1.0 AS x#654, abc AS y#655, 2022-05-02 17:32:56.606 AS z#656]                                                          :        +- Sort [x#654 ASC NULLS FIRST, y#655 ASC NULLS FIRST, x#654 ASC NULLS FIRST], true
!:              +- Filter ((amountPaid#664 > cast(100.0 as double)) AND (1 = cast(1 as int)))                                                                                                 :           +- Project [customerId#663, amountPaid#664 AS amount#653, 1.0 AS x#654, abc AS y#655, 2022-05-02 17:32:56.606 AS z#656]
!:                 +- HiveTableRelation [`default`.`sales`, org.apache.hadoop.hive.serde2.lazy.LazySimpleSerDe, Data Cols: [id#662, customerId#663, amountPaid#664], Partition Cols: []]      :              +- Filter ((amountPaid#664 > cast(100.0 as double)) AND (1 = cast(1 as int)))
!+- Project [customerId#666]                                                                                                                                                                  :                 +- HiveTableRelation [`default`.`sales`, org.apache.hadoop.hive.serde2.lazy.LazySimpleSerDe, Data Cols: [id#662, customerId#663, amountPaid#664], Partition Cols: []]
!   +- HiveTableRelation [`default`.`sales`, org.apache.hadoop.hive.serde2.lazy.LazySimpleSerDe, Data Cols: [id#665, customerId#666, amountPaid#667], Partition Cols: []]                     +- Project [customerId#666]
!                                                                                                                                                                                                +- HiveTableRelation [`default`.`sales`, org.apache.hadoop.hive.serde2.lazy.LazySimpleSerDe, Data Cols: [id#665, customerId#666, amountPaid#667], Partition Cols: []]
           
22/05/02 17:32:56 WARN PlanChangeLogger: 
=== Applying Rule org.apache.spark.sql.catalyst.optimizer.ReplaceDistinctWithAggregate ===
!Distinct                                                                                                                                                                                     Aggregate [customerId#663], [customerId#663]
 +- Join LeftAnti, (customerId#663 <=> customerId#666)                                                                                                                                        +- Join LeftAnti, (customerId#663 <=> customerId#666)
!   :- Distinct                                                                                                                                                                                  :- Aggregate [customerId#663], [customerId#663]
    :  +- Project [customerId#663]                                                                                                                                                               :  +- Project [customerId#663]
    :     +- Filter (customerId#663 < 20)                                                                                                                                                        :     +- Filter (customerId#663 < 20)
    :        +- Sort [x#654 ASC NULLS FIRST, y#655 ASC NULLS FIRST, x#654 ASC NULLS FIRST], true                                                                                                 :        +- Sort [x#654 ASC NULLS FIRST, y#655 ASC NULLS FIRST, x#654 ASC NULLS FIRST], true
    :           +- Project [customerId#663, amountPaid#664 AS amount#653, 1.0 AS x#654, abc AS y#655, 2022-05-02 17:32:56.606 AS z#656]                                                          :           +- Project [customerId#663, amountPaid#664 AS amount#653, 1.0 AS x#654, abc AS y#655, 2022-05-02 17:32:56.606 AS z#656]
    :              +- Filter ((amountPaid#664 > cast(100.0 as double)) AND (1 = cast(1 as int)))                                                                                                 :              +- Filter ((amountPaid#664 > cast(100.0 as double)) AND (1 = cast(1 as int)))
    :                 +- HiveTableRelation [`default`.`sales`, org.apache.hadoop.hive.serde2.lazy.LazySimpleSerDe, Data Cols: [id#662, customerId#663, amountPaid#664], Partition Cols: []]      :                 +- HiveTableRelation [`default`.`sales`, org.apache.hadoop.hive.serde2.lazy.LazySimpleSerDe, Data Cols: [id#662, customerId#663, amountPaid#664], Partition Cols: []]
    +- Project [customerId#666]                                                                                                                                                                  +- Project [customerId#666]
       +- HiveTableRelation [`default`.`sales`, org.apache.hadoop.hive.serde2.lazy.LazySimpleSerDe, Data Cols: [id#665, customerId#666, amountPaid#667], Partition Cols: []]                        +- HiveTableRelation [`default`.`sales`, org.apache.hadoop.hive.serde2.lazy.LazySimpleSerDe, Data Cols: [id#665, customerId#666, amountPaid#667], Partition Cols: []]

=== Applying Rule org.apache.spark.sql.catalyst.optimizer.PushDownPredicates ===
 Aggregate [customerId#663], [customerId#663]                                                                                                                                                 Aggregate [customerId#663], [customerId#663]
 +- Join LeftAnti, (customerId#663 <=> customerId#666)                                                                                                                                        +- Join LeftAnti, (customerId#663 <=> customerId#666)
    :- Aggregate [customerId#663], [customerId#663]                                                                                                                                              :- Aggregate [customerId#663], [customerId#663]
    :  +- Project [customerId#663]                                                                                                                                                               :  +- Project [customerId#663]
!   :     +- Filter (customerId#663 < 20)                                                                                                                                                        :     +- Sort [x#654 ASC NULLS FIRST, y#655 ASC NULLS FIRST, x#654 ASC NULLS FIRST], true
!   :        +- Sort [x#654 ASC NULLS FIRST, y#655 ASC NULLS FIRST, x#654 ASC NULLS FIRST], true                                                                                                 :        +- Project [customerId#663, amountPaid#664 AS amount#653, 1.0 AS x#654, abc AS y#655, 2022-05-02 17:32:56.606 AS z#656]
!   :           +- Project [customerId#663, amountPaid#664 AS amount#653, 1.0 AS x#654, abc AS y#655, 2022-05-02 17:32:56.606 AS z#656]                                                          :           +- Filter (((amountPaid#664 > cast(100.0 as double)) AND (1 = cast(1 as int))) AND (customerId#663 < 20))
!   :              +- Filter ((amountPaid#664 > cast(100.0 as double)) AND (1 = cast(1 as int)))                                                                                                 :              +- HiveTableRelation [`default`.`sales`, org.apache.hadoop.hive.serde2.lazy.LazySimpleSerDe, Data Cols: [id#662, customerId#663, amountPaid#664], Partition Cols: []]
!   :                 +- HiveTableRelation [`default`.`sales`, org.apache.hadoop.hive.serde2.lazy.LazySimpleSerDe, Data Cols: [id#662, customerId#663, amountPaid#664], Partition Cols: []]      +- Project [customerId#666]
!   +- Project [customerId#666]                                                                                                                                                                     +- HiveTableRelation [`default`.`sales`, org.apache.hadoop.hive.serde2.lazy.LazySimpleSerDe, Data Cols: [id#665, customerId#666, amountPaid#667], Partition Cols: []]
!      +- HiveTableRelation [`default`.`sales`, org.apache.hadoop.hive.serde2.lazy.LazySimpleSerDe, Data Cols: [id#665, customerId#666, amountPaid#667], Partition Cols: []]                  
           
22/05/02 17:32:56 WARN PlanChangeLogger: 
=== Applying Rule org.apache.spark.sql.catalyst.optimizer.FoldablePropagation ===
 Aggregate [customerId#663], [customerId#663]                                                                                                                                              Aggregate [customerId#663], [customerId#663]
 +- Join LeftAnti, (customerId#663 <=> customerId#666)                                                                                                                                     +- Join LeftAnti, (customerId#663 <=> customerId#666)
    :- Aggregate [customerId#663], [customerId#663]                                                                                                                                           :- Aggregate [customerId#663], [customerId#663]
    :  +- Project [customerId#663]                                                                                                                                                            :  +- Project [customerId#663]
!   :     +- Sort [x#654 ASC NULLS FIRST, y#655 ASC NULLS FIRST, x#654 ASC NULLS FIRST], true                                                                                                 :     +- Sort [1.0 ASC NULLS FIRST, abc ASC NULLS FIRST, 1.0 ASC NULLS FIRST], true
    :        +- Project [customerId#663, 1.0 AS x#654, abc AS y#655]                                                                                                                          :        +- Project [customerId#663, 1.0 AS x#654, abc AS y#655]
    :           +- Filter (((amountPaid#664 > cast(100.0 as double)) AND (1 = cast(1 as int))) AND (customerId#663 < 20))                                                                     :           +- Filter (((amountPaid#664 > cast(100.0 as double)) AND (1 = cast(1 as int))) AND (customerId#663 < 20))
    :              +- HiveTableRelation [`default`.`sales`, org.apache.hadoop.hive.serde2.lazy.LazySimpleSerDe, Data Cols: [id#662, customerId#663, amountPaid#664], Partition Cols: []]      :              +- HiveTableRelation [`default`.`sales`, org.apache.hadoop.hive.serde2.lazy.LazySimpleSerDe, Data Cols: [id#662, customerId#663, amountPaid#664], Partition Cols: []]
    +- Project [customerId#666]                                                                                                                                                               +- Project [customerId#666]
       +- HiveTableRelation [`default`.`sales`, org.apache.hadoop.hive.serde2.lazy.LazySimpleSerDe, Data Cols: [id#665, customerId#666, amountPaid#667], Partition Cols: []]                     +- HiveTableRelation [`default`.`sales`, org.apache.hadoop.hive.serde2.lazy.LazySimpleSerDe, Data Cols: [id#665, customerId#666, amountPaid#667], Partition Cols: []]

22/05/02 17:32:56 WARN PlanChangeLogger: 
=== Applying Rule org.apache.spark.sql.catalyst.optimizer.ConstantFolding ===
 Aggregate [customerId#663], [customerId#663]                                                                                                                                              Aggregate [customerId#663], [customerId#663]
 +- Join LeftAnti, (customerId#663 <=> customerId#666)                                                                                                                                     +- Join LeftAnti, (customerId#663 <=> customerId#666)
    :- Aggregate [customerId#663], [customerId#663]                                                                                                                                           :- Aggregate [customerId#663], [customerId#663]
    :  +- Project [customerId#663]                                                                                                                                                            :  +- Project [customerId#663]
    :     +- Sort [1.0 ASC NULLS FIRST, abc ASC NULLS FIRST, 1.0 ASC NULLS FIRST], true                                                                                                       :     +- Sort [1.0 ASC NULLS FIRST, abc ASC NULLS FIRST, 1.0 ASC NULLS FIRST], true
    :        +- Project [customerId#663, 1.0 AS x#654, abc AS y#655]                                                                                                                          :        +- Project [customerId#663, 1.0 AS x#654, abc AS y#655]
!   :           +- Filter (((amountPaid#664 > cast(100.0 as double)) AND (1 = cast(1 as int))) AND (customerId#663 < 20))                                                                     :           +- Filter (((amountPaid#664 > 100.0) AND true) AND (customerId#663 < 20))
    :              +- HiveTableRelation [`default`.`sales`, org.apache.hadoop.hive.serde2.lazy.LazySimpleSerDe, Data Cols: [id#662, customerId#663, amountPaid#664], Partition Cols: []]      :              +- HiveTableRelation [`default`.`sales`, org.apache.hadoop.hive.serde2.lazy.LazySimpleSerDe, Data Cols: [id#662, customerId#663, amountPaid#664], Partition Cols: []]
    +- Project [customerId#666]                                                                                                                                                               +- Project [customerId#666]
       +- HiveTableRelation [`default`.`sales`, org.apache.hadoop.hive.serde2.lazy.LazySimpleSerDe, Data Cols: [id#665, customerId#666, amountPaid#667], Partition Cols: []]                     +- HiveTableRelation [`default`.`sales`, org.apache.hadoop.hive.serde2.lazy.LazySimpleSerDe, Data Cols: [id#665, customerId#666, amountPaid#667], Partition Cols: []]

```

作业三

别名重命名示例

```scala
case class RenameAlias(spark: SparkSession) extends Rule[LogicalPlan] {
  override def apply(plan: LogicalPlan): LogicalPlan = {
    logInfo("In RenameAlias")
    plan transformAllExpressions {
      case e @ Alias(_, name) if name.startsWith("rename") => {
        logInfo("RenameAlias applied")
        e.withName(s"my_${e.name}")
      }
    }
  }
}

class MySparkSessionExtension extends (SparkSessionExtensions => Unit) {
  override def apply(extensions: SparkSessionExtensions): Unit = {
    extensions.injectResolutionRule { session =>
      RenameAlias(session)
    }
  }
}
```

`maven`打包后运行`spark-sql`

```shell
bin/spark-sql --jars week8-homework-1.0-SNAPSHOT.jar --conf spark.sql.extensions=org.geektime.time.exercise3.MySparkSessionExtension
```

执行如下查询验证结果：

```sparksql
set spark.sql.planChangeLog.level=WARN;
explain
select 1 as rename_name;
```

日志片段

```shell
=== Applying Rule org.geektime.time.exercise3.RenameAlias ===
!Project [1 AS rename_name#10]   Project [1 AS my_rename_name#10]
 +- OneRowRelation               +- OneRowRelation

```