Spark 2.x与1.x的对比及分析

阳光穿透心脏的1/2处 2024-04-19 13:19 82阅读 0赞

*  DataFrame与Dataset 统一化了，只剩下DataSet了
 *  flatMapToPair 由reture list变为reture  iterator
 *  ForeachRDD 不再return null
 *  更新状态的函数中,使用的Optional来自com.google.common.base，函数不能用
 *  Spark streaming 中JavaStreamingContextFactory类 废弃
 *  Kafka 只提供direct方式

## Spark 2.x 与 1.x 对比 ##

*  Spark 1.x：Spark Core（RDD）、Spark SQL（SQL+Dataframe+Dataset）、Spark Streaming、Spark MLlib、Spark Graphx
 *  Spark 2.x：Spark Core（RDD）、Spark SQL（ANSI-SQL+Subquery+Dataframe/Dataset）、Spark Streaming、Structured Streaming、Spark MLlib（Dataframe/Dataset）、Spark Graphx、Second Generation Tungsten Engine（Whole-stage code generation+Vectorization）

Spark 1.x 到 Spark 2.x，完全是一脉相承的关系，即 **Spark 2.x 基本上是基于 Spark 1.x 进行了更多的功能和模块的扩展，以及底层性能的改良**。绝对不是说，Spark 2.x 彻底淘汰和替代了 Spark 1.x 中的组件。而且实际上，对于 Spark 1.x 中 90% 以上的东西，Spark 2.x 几乎都完全保留了支持和延续，并没有做任何改变。这是必须要了解的一件事情。  
下面就对 Spark 2.x 中的每个组件都进行分析，了解组件的基本原理，以及其适用和不适用的场景。

## Spark 2.x 各组件分析 ##

### Spark Core（RDD） ###

从 Spark 诞生之日开始，RDD 就是 Spark 最主要的编程接口，重要程度类似于 Hadoop 中的 MapReduce。RDD，简单来说，就是一个不可变的分布式数据集，被分为多个 partition 从而在一个集群上分布式地存储。我们可以使用 RDD 提供的各种 transformation 和 action 算子，对 RDD 执行分布式的计算操作。  
Spark 2.0 开始，包括 Structured Streaming、Spark MLlib、Spark SQL 底层都开始基于 Dataframe/Dataset 来作为基础计算引擎，那么 Spark Core/RDD 是不是就要被淘汰了？回答是：不会！  
Spark 官方社区对于这个问题也是这个态度，Spark Core 绝对不会被淘汰掉。因为 Spark Core/RDD 作为一种 low-level 的 API 有它的较为底层的应用场景，虽然后续这种场景会越来越少，Dataframe/Dataset API 会逐渐替代原先 Spark Core 的一些场景，但是不可否认的是，这种场景还是存在的。此外，Dataframe/Dataset 实际上底层也是基于 Spark Core/RDD 构建的。所以说，Spark Core/RDD 是 Spark 生态中，不可替代的基础 API 和引擎，其他所有的组件几乎都是构建在它之上。未来它不会被淘汰，只是应用场景会减少而已。  
Spark 2.x 中，在离线批处理计算中，编程 API，除了 RDD 以外，还增强了 Dataframe/Dataset API。那么，我们到底什么时候应该使用 Spark Core/RDD 来进行编程呢？实际上，RDD 和 Dataset 最大的不同在于，RDD 是底层的 API 和内核，Dataset 实际上基于底层的引擎构建的 high-level 的计算引擎。

1.  如果我们需要对数据集进行非常底层的掌控和操作，比如说，手动管理 RDD 的分区，或者根据 RDD 的运行逻辑来结合各种参数和编程来进行较为底层的调优。因为实际上 Dataframe/Dataset 底层会基于 whole-stage code generation 技术自动生成很多代码，那么就意味着，当我们在进行线上报错的 troubleshooting 以及性能调优时，对 Spark 的掌控能力就会降低。而使用 Spark Core/RDD，因为其运行完全遵循其源码，因此我们完全可以在透彻阅读 Spark Core 源码的基础之上，对其进行 troubleshooting 和底层调优。（最重要的一点）
2.  我们要处理的数据是非结构化的，比如说多媒体数据，或者是普通文本数据。
3.  我们想要使用过程式编程风格来处理数据，而不想使用 domain-specific language 的编程风格来处理数据。
4.  我们不关心数据的 schema，即元数据。
5.  我们不需要 Dataframe/Dataset 底层基于的第二代 tungsten 引擎提供的 whole-stage code generation 等性能优化技术。

### Spark SQL（ANSI-SQL+Subquery） ###

Spark 2.x 一方面对 SQL 的支持做了大幅度的增强，另一方面，也通过优化了底层的计算引擎（第二代 tungsten 引擎，whole-stage code generation 等），提升了 SQL 的执行性能以及稳定性。  
所以在 Spark 2.x 中，一方面，开始鼓励大家多使用 Spark SQL 的 SQL 支持，采用 Spark SQL 来编写 SQL 进行最常见的大数据统计分析。比如可以尝试将 Hive 中的运行的一些 SQL 语句慢慢迁移到 Spark SQL 上来。另外一方面，一般一个新的大版本，都是不太稳定的，因此 Spark SQL 虽然在功能、性能和稳定性上做了很多的增强，但是难免还是会有很多的坑。因此建议在做 Hive/RDBMS（比如 Oracle）到 Spark SQL 的迁移时，要小心谨慎，一点点迁移，同时做好踩坑的准备。

### Spark SQL（Dataframe/Dataset） ###

就像 RDD 一样，Dataframe 也代表一个不可变的分布式数据集。与 RDD 不同的一点是，Dataframe 引入了 schema 的概念，支持以复杂的类型作为元素类型，同时指定 schema，比如 Row。因此 Dataframe 更像是传统关系型数据库中的表的概念。为了提升开发人员对大数据的处理能力，Dataframe 除了提供 schema 的引入，还基于 Schema 提供了很多 RDD 所不具备的 high-level API，以及一些 domain-specific language（特定领域编程语言）。**但是在 Spark 2.0 中，Dataframe 和 Dataset 合并了**，Dataframe 已经不是一个单独的概念了，目前仅仅只是 Dataset\[Row\] 的一个类型别名而已，你可以理解为 Dataframe 就是 Dataset。

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3NpbmF0XzI2ODExMzc3_size_16_color_FFFFFF_t_70][]

从 Spark 2.0 开始，Dataset 有两种表现形式：typed API 和 untyped API。我们可以认为，Dataframe 就是 Dataset\[Row\] 的别名，Row 就是一个 untyped 类型的对象，因为 Row 是类似于数据库中的一行，我们只知道里面有哪些列，但是有些列即使不存在，我们也可以这对这些不存在的列进行操作。因此其被定义为 untyped，就是弱类型。  
而 Dataset\[T\] 本身，是一种 typed 类型的 API，其中的 Object 通常都是我们自己自定义的 typed 类型的对象，因为对象是我们自己定义的，所以包括字段命名以及字段类型都是强类型的。目前 Scala 支持 Dataset 和 Dataframe 两种类型，Java 仅仅支持 Dataset 类型，Python 和 R 因为不具备 compile-time type-safety 特性，因此仅仅支持 Dataframe。

Dataset API 有哪些优点

*  **静态类型以及运行时的类型安全性**  
    SQL 语言具有最不严格的限制，而 Dataset 具有最严格的限制。SQL 语言在只有在运行时才能发现一些错误，比如类型错误，但是由于 Dataframe/Dataset 目前都是要求类型指定的（静态类型），因此在编译时就可以发现类型错误，并提供运行时的类型安全。比如说，如果我们调用了一个不属于 Dataframe 的 API，编译时就会报错。但是如果你使用了一个不存在的列，那么也只能到运行时才能发现了。而最严格的就是 Dataset 了，因为 Dataset 是完全基于 typed API 来设计的，类型都是严格而且强类型的，因此如果你使用了错误的类型，或者对不存在的列进行了操作，都能在编译时就发现。

<table> 
 <thead> 
  <tr> 
   <th>错误类型</th> 
   <th>SQL</th> 
   <th>Dataframe</th> 
   <th>Dataset</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>Syntax Error</td> 
   <td>Runtime</td> 
   <td>Compile Time</td> 
   <td>Compile Time</td> 
  </tr> 
  <tr> 
   <td>Analysis Error</td> 
   <td>Runtime</td> 
   <td>Runtime</td> 
   <td>Compile Time</td> 
  </tr> 
 </tbody> 
</table>

*  **将半结构化的数据转换为 typed 自定义类型**  
    举例来说，如果我们现在有一份包含了学校中所有学生的信息，是以 JSON 字符串格式定义的，比如：\{“name”: “leo”, “age”, 19, “classNo”: 1\}。我们可以自己定义一个类型，比如 case class Student(name: String, age: Integer, classNo: Integer)。接着我们就可以加载指定的 json 文件，并将其转换为 typed 类型的 Dataset\[Student\]，比如 val ds = spark.read.json("students.json").as\[Student\]。  
    在这里，Spark 会执行三个操作：  
    1）. Spark 首先会读取 json 文件，并且自动推断其 schema，然后根据 schema 创建一个 Dataframe。  
    2）. 在这里，会创建一个 Dataframe=Dataset\[Row\]，使用 Row 来存放你的数据，因为此时还不知道具体确切的类型。  
    3）. 接着将 Dataframe 转换为 Dataset\[Student\]，因为此时已经知道具体的类型是 Student 了。  
    这样，我们就可以将半结构化的数据，转换为自定义的 typed 结构化强类型数据集。并基于此，得到之前说的编译时和运行时的类型安全保障。
 *  **API 的易用性**  
    Dataframe/Dataset 引入了很多的 high-level API，并提供了 domain-specific language 风格的编程接口。这样的话，大部分的计算操作，都可以通过 Dataset 的 high-level API 来完成。通过 typed 类型的 Dataset，我们可以轻松地执行 agg、select、sum、avg、map、filter、groupBy 等操作。使用 domain-specific language 也能够轻松地实现很多计算操作，比如类似 RDD 算子风格的 map()、filter() 等。
 *  **性能**  
    除了上述的优点，Dataframe/Dataset 在性能上也有很大的提升。首先，Dataframe/Dataset 是构建在 Spark SQL 引擎之上的，它会根据你执行的操作，使用 Spark SQL 引擎的 Catalyst 来生成优化后的逻辑执行计划和物理执行计划，可以大幅度节省内存或磁盘的空间占用的开销（相对于 RDD 来说，Dataframe/Dataset 的空间开销仅为 1/3~1/4），也能提升计算的性能。其次，Spark 2.x 还引入第二代 Tungsten 引擎，底层还会使用 whole-stage code generation、vectorization 等技术来优化性能。

**什么时候应该使用 Dataframe/Dataset，而不是 RDD 呢？**

1.  如果需要更加丰富的计算语义，high-level 的抽象语义，以及 domain-specific API。
2.  如果计算逻辑需要 high-level 的 expression、filter、map、aggregation、average、sum、SQL、列式存储、lambda 表达式等语义，来处理半结构化，或结构化的数据。
3.  如果需要高度的编译时以及运行时的类型安全保障。
4.  如果想要通过 Spark SQL 的 Catalyst 和 Spark 2.x 的第二代 Tungsten 引擎来提升性能。
5.  如果想要通过统一的 API 来进行离线、流式、机器学习等计算操作。
6.  如果是 R 或 Python 的用户，那么只能使用 Dataframe。

最后，实际上，Spark 官方社区对 RDD 和 Dataframe/Dataset 的建议时，按照各自的特点，根据的需求场景，来灵活的选择最合适的引擎。甚至说，在一个 Spark 应用中，也可以将两者结合起来一起使用。

### Spark Streaming&Structured Streaming ###

Spark Streaming 是老牌的 Spark 流式计算引擎，底层基于 RDD 计算引擎。除了类似 RDD 风格的计算 API 以外，也提供了更多的流式计算语义，比如 window、updateStateByKey、transform 等。同时对于流式计算中重要的数据一致性、容错性等也有一定的支持。  
但是 Spark 2.x 中也推出了全新的基于 Dataframe/Dataset 的 Structured Streaming 流式计算引擎。相较于 Spark Streaming 来说，其最大的不同之处在于，采用了全新的逻辑模型，提出了 real-time incremental table 的概念，更加统一了流式计算和离线计算的概念，减轻了用户开发的负担。同时还提供了（可能在未来提供）高度封装的特性，比如双流的全量 join、与离线数据进行 join 的语义支持、内置的自动化容错机制、内置的自动化的一次且仅一次的强一致性语义、time-based processing、延迟数据达到的自动处理、与第三方外部存储进行整合的 sink 概念，等等高级特性。大幅度降低了流式计算应用的开发成本。  
这里要提的一句是，首先，目前暂时建议使用 Spark Streaming，因为 Spark Streaming 基于 RDD，而且经过过个版本的考验，已经趋向于稳定。对于 Structured Streaming 来说，一定要强调，在 Spark 2.0 版本刚推出的时候，千万别在生产环境使用，因为目前官方定义为 beta 版，就是测试版，里面可能有很多的 bug 和问题，而且上述的各种功能还不完全，很多功能还没有。因此 Structured Streaming 的设计理念虽然非常好，但是可以在后续的版本中再考虑使用。目前可以保持关注和学习，并做一些实验即可。

### Spark MLlib&GraphX ###

Spark MLlib 未来将主要基于 Dataframe/Dataset API 来开发。而且还会提供更多的机器学习算法。因此可以主要考虑使用其 spark.ml 包下的 API 即可。  
Spark GraphX，目前发展较为缓慢，如果有图计算相关的应用，可以考虑使用。

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3NpbmF0XzI2ODExMzc3_size_16_color_FFFFFF_t_70]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/18/f6bcf1a59571453388b60d7e010f7c33.png