spark 2.0主要特性预览

 

2016-05-29 朱洁 hadoop技术学习

 

spark 2.0相比老版本变化很大，已经发布了预览版本。原始的英文版databricks的博客：https://databricks.com/blog/2016/05/11/apache-spark-2-0-technical-preview-easier-faster-and-smarter.html

 

变化点主要有：

1、统一api 到datasets

DataFrame 和 Dataset 的功能是什么?它们都是提供给用户使用，包括各类操作接口的 API。1.3 版本引入 DataFrame，1.6 版本引入 Dataset，2.0 提供的功能是将二者统一，即保留 Dataset，而把 DataFrame 定义为 Dataset[Row]，即是 Dataset 里的元素对象为 Row 的一种(SPARK-13485)。

DataFrame，它就是提供了一系列操作 API，与 RDD API 相比较，DataFrame 里操作的数据都是带有 Schema 信息，所以 DataFrame 里的所有操作是可以享受 Spark SQL Catalyst optimizer 带来的性能提升，比如 code generation 以及 Tungsten 等。执行过程如下图所示：

 

 

但是 DataFrame 出来后发现有些情况下 RDD 可以表达的逻辑用 DataFrame 无法表达。比如 要对 group by 或 join 后的结果用自定义的函数,可能用 SQL 是无法表达的。如下代码：

case class ClassData(a: String, b: Int)

case class ClassNullableData(a: String, b: Integer)

val ds = Seq(ClassData("a", 1), ClassData("a", 2)).toDS()

val agged = ds.groupByKey(d => ClassNullableData(d.a, null))

.mapGroups {

case (key, values) => key.a + values.map(_.b).sum

}

中间处理过程的数据是自定义的类型,并且 groupby 后的聚合逻辑也是自定义的，故用 SQL 比较难以表达，所以提出了 Dataset API。Dataset API 扩展 DataFrame API 支持静态类型和运行已经存在的 Scala 或 Java 语言的用户自定义函数。同时 Dataset 也能享受 Spark SQL 里所有性能 带来的提升。那么后面发现 Dataset 是包含了 DataFrame 的功能，这样二者就出现了很大的冗余，故在 2.0 时将二者统一，保留 Dataset API，把 DataFrame 表示为 Dataset[Row]，即 Dataset 的子集。

因此我们在使用 API 时，优先选择 DataFrame & Dataset，因为它的性能很好，而且以后的优化它都可以享受到，但是为了兼容早期版本的程序，RDD API 也会一直保留着。后续 Spark 上层的库将全部会用 DataFrame，比如 MLlib、Streaming、Graphx 等。

 

2、全流程code generation

我们看其中一个例子：

select count(*) from store_sales where ss_item_sk = 1000

那么在翻译成计算引擎的执行计划如下图：

 

 

 

而通常物理计划的代码是这样实现的：

class Filter {

def next(): Boolean = {

var found = false

while (!found && child.next()) {

found = predicate(child.fetch())

}

return found

}

def fetch(): InternalRow = {

child.fetch()

}...

}

但是真正如果我们用 hard code 写的话，代码是这样的：

var count = 0

for (ss_item_sk in store_sales) {

if (ss_item_sk == 1000) {

count += 1

}

}

发现二者相关如下图所示：

 

 

 

那么如何使得计算引擎的物理执行速度能达到 hard code 的性能呢？这就提出了 whole-stage code generation，即对物理执行的多次调用转换为代码 for 循环，类似 hard code 方式，减少中间执行的函数调用次数，当数据记录多时，这个调用次数是很大。 最后这个优化带来的性能提升如下图所示：

 

 

 

从 benchmark 的结果可以看出，使用了该特性后各操作的性能都有很大的提升。

tpc-ds的对比测试结果也非常好（spark 1.6对比spark 2.0） ：

 

 

 

 

 

 

 

tpc-ds测试的效果，除流全流程的code generation，还有大量在优化器的优化如空值传递以及对parquet扫描的3倍优化

 

 

3、抛弃Dstrem API，新增结构化流api

Spark Streaming 是把流式计算看成一个一个的离线计算来完成流式计算，提供了一套 Dstream 的流 API，相比于其他的流式计算，Spark Streaming 的优点是容错性和吞吐量上要有优势，关于 Spark Streaming 的详细设计思想和分析，可以到 https://github.com/lw-lin/CoolplaySpark 进行详细学习和了解。

在 2.0 以前的版本，用户在使用时，如果有流计算，又有离线计算，就需要用二套 API 去编写程序，一套是 RDD API，一套是 Dstream API。而且 Dstream API 在易用性上远不如 SQL 或 DataFrame。

为了真正将流式计算和离线计算在编程 API 上统一，同时也让 Streaming 作业能够享受 DataFrame/Dataset 上所带来的优势：性能提升和 API 易用，于是提出了 Structured Streaming。最后我们只需要基于 DataFrame/Dataset 可以开发离线计算和流式计算的程序，很容易使得 Spark 在 API 跟业界所说的 DataFlow 来统一离线计算和流式计算效果一样。

比如在做 Batch Aggregation 时我们可以写成下面的代码：

 

 

那么对于流式计算时，我们仅仅是调用了 DataFrame/Dataset 的不同函数代码，如下：

 

 

最后，在 DataFrame/Dataset 这个 API 上可以完成如下图所示的所有应用：

 

 

4、最后 2.0 版本还有一些其他的特性，如：

1. 用 SparkSession 替换掉原来的 SQLContext and HiveContext。

2. mllib 里的计算用 DataFrame-based API 代替以前的 RDD 计算逻辑。

3. 提供更多的分布式R 语言算法。

4. 支持ml pipeline持久化

5. 更简单，更高性能的Accumulator API

 

 

微信扫一扫
关注该公众号