Spark的RDD原理以及2.0特性的介绍(3)

但是真正如果我们用 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 的结果可以看出,使用了该特性后各操作的性能都有很大的提升.

Structured Streaming

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 上可以完成如下图所示的所有应用：

其他主要性能提升

1. 采用 vectorized Parquet decoder 读取 parquet 上数据.以前是一行一行的读取,然后处理.现在改为一次读取 4096 行记录,不需要每处理一行记录去调用一次 Parquet 获取记录的方法,而是改为一批去调用一次(SPARK-12854).加上 Parquet 本身是列式存储,这个优化使得 Parquet 读取速度提高 3 倍.
2. 采有 radix sort 来提高 sort 的性能(SPARK-14724).在某些情况下排序性能可以提高 10-20 倍.
3. 使用 VectorizedHashmap 来代替 Java 的 HashMap 加速 group by 的执行(SPARK-14319).
4. 将 Hive 中的 Window 函数用 Native Spark Window 实现,因为 Native Spark Window 在内存管理上有优势(SPARK-8641).
5. 避免复杂语句中的逻辑相同部分在执行时重复计算(SPARK-13523).
6. 压缩算法默认使用 LZ4(SPARK-12388).

语句的增强

1. 建立新的语法解析(SPARK-12362)满足所有的 SQL 语法,这样即合并 Hive 和标准 SQL 的语句解析,同时不依赖 Hive 的语法解析 jar(SPARK-14776).之前版本二者的语法解析是独立的,这样导致在标准 SQL 中无法使用窗口函数或者 Hive 的语法,而在使用 Hive 语法时无法使用标准 SQL 的语法,比如 In/Exists 子句等.在 SQL 编写时,没法在一个 Context 把二者的范围全部支持,然而有了这个特性后,使得 SQL 语句表达更强大,后续要增加任何语法,只需要维护这一个语法解析即可.当然缺点是后续 Hive 版本的新语法,需要手动添加进来.
2. 支持 intersect/except(SPARK-12542).如 select * from t1 except select * from t2 或者 select * from t1 intersect select * from t2.
3. 支持 uncorrelated scalar subquery(SPARK-13417).如 select (select min(value) from testData where key = (select max(key) from testData) – 1).
4. 支持 DDL/DML(SPARK-14118).之前 DDL/DML 语句是调用 Hive 的 DDL/DML 语句命令来完成,而现在是直接在 Spark SQL 上就可以完成.
5. 支持 multi-insert(SPARK-13924).
6. 支持 exist(SPARK-12545)和 NOT EXISTS(SPARK-10600),如 select * from (select 1 as a union all select 2 as a) t where exists (select * from (select 1 as b) t2 where b = a and b < 2).
7. 支持 subqueries 带有 In/Not In 子句(SPARK-4226),如 select * from (select 1 as a union all select 2 as a) t where a in (select b as a from t2 where b < 2).
8. 支持 select/where/having 中使用 subquery(SPARK-12543),如 select * from t where a = (select max(b) from t2) 或 select max(a) as ma from t having ma = (select max(b) from t2).
9. 支持 LeftSemi/LeftAnti(SPARK-14853).
10. 支持在条件表达式 In/Not In 里使用子句(SPARK-14781),如 select * from l where l.a in (select c from r) or l.a in (select c from r where l.b < r.d).
11. 支持所有的 TPCDS 语句(SPARK-12540).

与以前版本兼容(SPARK-11806)

1. 不支持运行在 Hadoop 版本 < 2.2 上(SPARK-11807).
2. 去掉 HTTPBroadcast(SPARK-12588).
3. 去掉 HashShuffleManager(SPARK-14667).
4. 去掉 Akka RPC.
5. 简化与完善 accumulators and task metrics(SPARK-14626).
6. 将 Hive 语法解析以及语法移至 Core 里(SPARK-14825),在没有 Hive 元数据库和 Hive 依赖包时,我们可以像之前版本使用标准 SQL 一样去使用 HiveQL 语句.

1.6 版本严重问题的解决

在 http://geek.csdn.net/news/detail/70162 提到的 1.6 问题中 Spillable 集合内存溢出问题在 SPARK-4452 里已解决,BlockManager 死锁问题在 SPARK-12757 里已解决.

（编辑：ASP站长网）