# final-homework

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

## Basic Information

- **Project Name**: final-homework
- **Description**: 极客时间大数据训练营最后一次作业
- **Primary Language**: Unknown
- **License**: Not specified
- **Default Branch**: master
- **Homepage**: None
- **GVP Project**: No

## Statistics

- **Stars**: 0
- **Forks**: 0
- **Created**: 2022-08-16
- **Last Updated**: 2022-08-20

## Categories & Tags

**Categories**: Uncategorized

**Tags**: None

## README

# 题目一

> ## 题目一: 分析一条 TPCDS SQL
>
> 分析一条 TPCDS SQL（请基于 Spark 3.1.1 版本解答）
>
> - 运行该 SQL，如 q38，并截图该 SQL 的 SQL 执行图
> - 该 SQL 用到了哪些优化规则（optimizer rules）
> - 请各用不少于 200 字描述其中的两条优化规则
> - SQL 从中任意选择一条：https://github.com/apache/spark/tree/master/sql/core/src/test/resources/tpcds
>
> 帮助文档：如何运行该 SQL：
>
> ```shell
> 1. 从 github 下载 TPCDS 数据生成器
> git clone https://github.com/maropu/spark-tpcds-datagen.git
> cd spark-tpcds-datagen
>  
> 2. 下载 Spark3.1.1 到 spark-tpcds-datagen 目录并解压
> wget https://archive.apache.org/dist/spark/spark-3.1.1/spark-3.1.1-bin-hadoop2.7.tgz
> tar -zxvf spark-3.1.1-bin-hadoop2.7.tgz
>  
> 3. 生成数据
> mkdir -p tpcds-data-1g
> export SPARK_HOME=./spark-3.1.1-bin-hadoop2.7
> ./bin/dsdgen --output-location tpcds-data-1g
>  
> 4. 下载三个 test jar 并放到当前目录
> wget https://repo1.maven.org/maven2/org/apache/spark/spark-catalyst_2.12/3.1.1/spark-catalyst_2.12-3.1.1-tests.jar
> wget https://repo1.maven.org/maven2/org/apache/spark/spark-core_2.12/3.1.1/spark-core_2.12-3.1.1-tests.jar
> wget https://repo1.maven.org/maven2/org/apache/spark/spark-sql_2.12/3.1.1/spark-sql_2.12-3.1.1-tests.jar
>  
> 5. 执行 SQL
> ./spark-3.1.1-bin-hadoop2.7/bin/spark-submit --class org.apache.spark.sql.execution.benchmark.TPCD
> ```

## 操作

操作上略有改动，目录结构为

```shell
➜ ls -lrt
total 8216
drwxr-xr-x@ 17 fuwuchen  staff      544  2 22  2021 spark-3.1.1-bin-hadoop2.7
drwxr-xr-x@ 14 fuwuchen  staff      448  6 30  2021 spark-tpcds-datagen-master
drwxr-xr-x  26 fuwuchen  staff      832  8 14 16:56 tpcds-data-1g
-rw-r--r--   1 fuwuchen  staff  3374473  8 14 17:15 result.log
-rw-r--r--@  1 fuwuchen  staff     6347  8 14 17:40 README.md
```

3 个 test jar 放置到 `spark-3.1.1-bin-hadoop2.7/jars` 目录下，并运行如下命令：

```shell
./spark-3.1.1-bin-hadoop2.7/bin/spark-submit --conf spark.sql.planChangeLog.level=WARN --class org.apache.spark.sql.execution.benchmark.TPCDSQueryBenchmark ./spark-3.1.1-bin-hadoop2.7/jars/spark-sql_2.12-3.1.1-tests.jar --data-location tpcds-data-1g --query-filter "q12" > result-12.log 2>&1
```

通过查找以下内容得到查询相关日志，在本例中为 7769 行开始

```txt
Running benchmark: TPCDS Snappy
  Running case: q12
```

通过如下命令查看 `result.log` 该 SQL 使用的优化规则

```shell
awk 'NR>7769' result-12.log | grep -o "org.apache.spark.sql.catalyst.optimizer.*" | awk '{print $1}' | sort | uniq -d
```

得到

```shell
org.apache.spark.sql.catalyst.optimizer.CollapseProject
org.apache.spark.sql.catalyst.optimizer.ColumnPruning
org.apache.spark.sql.catalyst.optimizer.ConstantFolding
org.apache.spark.sql.catalyst.optimizer.DecimalAggregates
org.apache.spark.sql.catalyst.optimizer.InferFiltersFromConstraints
org.apache.spark.sql.catalyst.optimizer.PushDownPredicates
org.apache.spark.sql.catalyst.optimizer.ReorderJoin
org.apache.spark.sql.catalyst.optimizer.RewritePredicateSubquery
org.apache.spark.sql.catalyst.optimizer.SimplifyCasts
```

执行图如下

![image-20220820081354515](https://testpicp.oss-cn-shanghai.aliyuncs.com/uPic/image-20220820081354515.png)

执行结果如下

![image-20220816150015815](https://testpicp.oss-cn-shanghai.aliyuncs.com/uPic/image-20220816150015815.png)

## 分析

原始 SQL

```sql
SELECT
  i_item_desc,
  i_category,
  i_class,
  i_current_price,
  sum(ws_ext_sales_price) AS itemrevenue,
  sum(ws_ext_sales_price) * 100 / sum(sum(ws_ext_sales_price))
  OVER
  (PARTITION BY i_class) AS revenueratio
FROM
  web_sales, item, date_dim
WHERE
  ws_item_sk = i_item_sk
    AND i_category IN ('Sports', 'Books', 'Home')
    AND ws_sold_date_sk = d_date_sk
    AND d_date BETWEEN cast('1999-02-22' AS DATE)
  AND (cast('1999-02-22' AS DATE) + INTERVAL 30 days)
GROUP BY
  i_item_id, i_item_desc, i_category, i_class, i_current_price
ORDER BY
  i_category, i_class, i_item_id, i_item_desc, revenueratio
LIMIT 100
```

其中 `web_sales` 表是事实表，`item` 表和 `date_dim` 表是维度表。

`web_sales` 表的定义为

![image-20220816094827543](https://testpicp.oss-cn-shanghai.aliyuncs.com/uPic/image-20220816094827543.png)

![image-20220816094844930](https://testpicp.oss-cn-shanghai.aliyuncs.com/uPic/image-20220816094844930.png)

`item` 表的定义为

![image-20220816093437036](https://testpicp.oss-cn-shanghai.aliyuncs.com/uPic/image-20220816093437036.png)

`date_dim` 表的定义为

![image-20220816094727002](https://testpicp.oss-cn-shanghai.aliyuncs.com/uPic/image-20220816094727002.png)

### 规则一：PushDownPredicates

该规则的作用是通过尽可能早地过滤数据，从而提升 `Spark`  的 `join` 性能。在查询中多次应用到该规则，以下面为例

下推前的计划：

![image-20220816102021746](https://testpicp.oss-cn-shanghai.aliyuncs.com/uPic/image-20220816102021746.png)

下推后的计划：

![image-20220816101941666](https://testpicp.oss-cn-shanghai.aliyuncs.com/uPic/image-20220816101941666.png)

`ws_sold_date_sk` 是 `web_sales` 的一个字段，该非空过滤条件在原始 SQL 中不存在，而是来源于 `InferFiltersFromConstraints` ，对应于 `date_dim` 表中 `d_date_sk` 定义不能为空。在应用 `PushDownPredicates` 规则前，该谓词在 `web_sales` 和 `date_dim` 表内连接之后，而应用规则后不仅下推到内连接前，并和已有的谓词进行了合并，最终只产生了一个 `Filter` 。

该行为在 `Spark 3.1.1`的代码中对应`spark-3.1.1/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/Optimizer.scala`：

```scala
/**
 * The unified version for predicate pushdown of normal operators and joins.
 * This rule improves performance of predicate pushdown for cascading joins such as:
 *  Filter-Join-Join-Join. Most predicates can be pushed down in a single pass.
 */
object PushDownPredicates extends Rule[LogicalPlan] with PredicateHelper {
  def apply(plan: LogicalPlan): LogicalPlan = plan transform {
    CombineFilters.applyLocally
      .orElse(PushPredicateThroughNonJoin.applyLocally)
      .orElse(PushPredicateThroughJoin.applyLocally)
  }
}
```

可见会根据是否有 `join` 应用不同的操作，并进行谓词的合并。对应有 `join` 的情况下提到了支持下推的 `join` 类型，并分别进行了实现：

```scala
/**
 * Pushes down [[Filter]] operators where the `condition` can be
 * evaluated using only the attributes of the left or right side of a join.  Other
 * [[Filter]] conditions are moved into the `condition` of the [[Join]].
 *
 * And also pushes down the join filter, where the `condition` can be evaluated using only the
 * attributes of the left or right side of sub query when applicable.
 *
 * Check https://cwiki.apache.org/confluence/display/Hive/OuterJoinBehavior for more details
 */
object PushPredicateThroughJoin extends Rule[LogicalPlan] with PredicateHelper {
  ...
    private def canPushThrough(joinType: JoinType): Boolean = joinType match {
    case _: InnerLike | LeftSemi | RightOuter | LeftOuter | LeftAnti | ExistenceJoin(_) => true
    case _ => false
  }
}
```

另一部分是没有 `join` 时能进行的下推，对应 `window`, `group by`, `union` 等情况，可查看 `PushPredicateThroughNonJoin` 对应的实现。

此外整个规则对支持下推的谓词做了约束，示例如下：

```scala
/**
 * Pushes [[Filter]] operators through many operators iff:
 * 1) the operator is deterministic
 * 2) the predicate is deterministic and the operator will not change any of rows.
 *
 * This heuristic is valid assuming the expression evaluation cost is minimal.
 */
object PushPredicateThroughNonJoin extends Rule[LogicalPlan] with PredicateHelper {
  ...
  val applyLocally: PartialFunction[LogicalPlan, LogicalPlan] = {
    // SPARK-13473: We can't push the predicate down when the underlying projection output non-
    // deterministic field(s).  Non-deterministic expressions are essentially stateful. This
    // implies that, for a given input row, the output are determined by the expression's initial
    // state and all the input rows processed before. In another word, the order of input rows
    // matters for non-deterministic expressions, while pushing down predicates changes the order.
    // This also applies to Aggregate.
    ... 
  }
```

可知算子和谓词需要是确定性的，指定输入有确定的输出。

### 规则二：RewritePredicateSubquery

该规则用于重写`exist`、`in`等查询为 `semi/anti join`。在原 SQL 的 `where` 子句中有 `i_category IN ('Sports', 'Books', 'Home')` 条件，此处使用的是固定常数因此和普通过滤条件并无太大区别，在应用规则前后逻辑计划没有改变。

实现对应源码 `spark-3.1.1/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/subquery.scala`

```scala
/**
 * This rule rewrites predicate sub-queries into left semi/anti joins. The following predicates
 * are supported:
 * a. EXISTS/NOT EXISTS will be rewritten as semi/anti join, unresolved conditions in Filter
 *    will be pulled out as the join conditions.
 * b. IN/NOT IN will be rewritten as semi/anti join, unresolved conditions in the Filter will
 *    be pulled out as join conditions, value = selected column will also be used as join
 *    condition.
 */
object RewritePredicateSubquery extends Rule[LogicalPlan] with PredicateHelper
```

实现中会将 `exist`，`in` 改写成 `left semi join`，`not exist`，`not in` 则改写成 `left anti join`，对应的值列表会变成 `join` 条件，并且会处理 `join` 条件冲突的情况（SPARK-21835、SPARK-26078）和空值等情况，对应注释

```scala
// SPARK-21835: It is possibly that the two sides of the join have conflicting attributes,
// the produced join then becomes unresolved and break structural integrity. We should
// de-duplicate conflicting attributes.
// SPARK-26078: it may also happen that the subquery has conflicting attributes with the outer
// values. In this case, the resulting join would contain trivially true conditions (eg.
// id#3 = id#3) which cannot be de-duplicated after. In this method, if there are conflicting
// attributes in the join condition, the subquery's conflicting attributes are changed using
// a projection which aliases them and resolves the problem.

...

// This is a NULL-aware (left) anti join (NAAJ) e.g. col NOT IN expr
// Construct the condition. A NULL in one of the conditions is regarded as a positive
// result; such a row will be filtered out by the Anti-Join 
// Note that will almost certainly be planned as a Broadcast Nested Loop join.
// Use EXISTS if performance matter
// Deduplicate conflicting attributes if any.
val newSub = dedupSubqueryOnSelfJoin(p, sub, Some(values))
val inConditions = values.zip(newSub.output).map(EqualTo.tupled)
val (joinCond, outerPlan) = rewriteExistentialExpr(inConditions, p)
// Expand the NOT IN expression with the NULL-aware semantic
// to its full form. That is from:
//   (a1,a2,...) = (b1,b2,...)
// to
//   (a1=b1 OR isnull(a1=b1)) AND (a2=b2 OR isnull(a2=b2)) AND ...
val baseJoinConds = splitConjunctivePredicates(joinCond.get)
val nullAwareJoinConds = baseJoinConds.map(c => Or(c, IsNull(c)))
// After that, add back the correlated join predicate(s) in the subquery
// Example:
// SELECT ... FROM A WHERE A.A1 NOT IN (SELECT B.B1 FROM B WHERE B.B2 = A.A2 AND B.B3 > 1)
// will have the final conditions in the LEFT ANTI as
// (A.A1 = B.B1 OR ISNULL(A.A1 = B.B1)) AND (B.B2 = A.A2) AND B.B3 > 1
```

### 参考

1. Apache Spark:https://github.com/maropu/spark-tpcds-datagen/blob/master/src/main/resources/tpcds/queries/q12.sql
2. TPCDS: https://www.tpc.org/tpc_documents_current_versions/pdf/tpc-ds_v3.2.0.pdf

# 题目二

> ## 题目二：架构设计题
>
> 你是某互联网公司的大数据平台架构师，请设计一套基于 Lambda 架构的数据平台架构，要求尽可能多的把课程中涉及的组件添加到该架构图中。并描述 Lambda 架构的优缺点，要求不少于 300 字。

![lambda.drawio2](https://testpicp.oss-cn-shanghai.aliyuncs.com/uPic/lambda.drawio2.png)

Lambda 架构以离线链路数据为准，实时链路数据为补充，能够应对海量数据的处理和增量数据的快速响应，是经典的大数据架构，其优点有：

- 实现简单：可以根据需要分别建设离线和实时链路，能够实现满足需求的通用架构

- 容错：由于离线链路有数据备份，且出错可以通过重跑任务纠正数据，即使实时链路计算有误，仍可以通过和离线链路对数从而得到准确的结果
- 可扩展：所有组件均为分布式组件，可通过横向扩展增加系统处理能力

缺点有：

- 复杂性和成本：由于需要分别维护离线和实时两条链路，组件多，需要的资源和人力成本高
- 不一致性：需要离线和实时链路实现相同的功能，可能造成统计口径不一致，进而无法产出有效信息
- 处理瓶颈：通常离线链路需要 T+1 频率触发计算，如果处理时间过长影响后续任务，可能导致整个系统的产出下降。

参考：

1. https://www.databricks.com/glossary/lambda-architecture
2. https://www.sohu.com/a/453919627_185201