转载

逆水行舟，看前行中的Spark

《程序员》电子刊为您提供最前沿的技术新知，让开发者在云计算、大数据、移动开发时代，成为业界领航者。您将掌握第一手业界动态、世界趋势，包括专家丰富而完整的倾囊相授和更深入的报道。如果您想投稿，请发送邮件至zhangzy#csdn.net（请把#改成@）。

过去两年，Spark以惊人的速度发展着，其主要表现在用户越来越多，社区愈加活跃，而更加丰富的生态更直接说明了Spark的魅力所在。同时，在用户数量和用户质量两个方面，Spark都给出了积极的信号。此外，在生态建设上，Spark同样取得了极大的成功，其主要体现在application、environment及data source三个方面。值得一提的是，截止到目前，Spark的贡献者已经超过650人，而在另一方面，围绕着Spark创业的公司同样随之增多，“Spark as a Service”这个概念也被越来越多的人接受，Spark的未来值得期待。

到目前为止，Spark看起来一帆风顺，但事实上，大数据这个领域从来不缺强者，其中最具代表性的无疑当属Flink（https://flink.apache.org）。Flink采用了MPP的思想，具备很多有意思的设计和特性。本文不准备用过多的篇幅来介绍Flink，主要给大家分享下Spark最近几个极其重要的改进。注意，下面所提到的改进有些已经实现，而有些尚未完成。

Project Tungsten

着眼Spark近期发展脚步，最引人瞩目的无疑当属钨丝计划（Project Tungsten）。 Kay Ousterhout在名为”Making Sense of Performance in Data Analytics Frameworks”的论文中谈到，类似Spark这样的计算框架，其瓶颈主要在于CPU与内存，而不是大家之前所认为的磁盘IO及网络开销。至于网络及磁盘IO为什么在很多情况下不是瓶颈的原因其实也很清晰，带宽增大、SSD或者磁盘阵列的使用都可以缓解这个问题，但是在序列化、反序列化及Hash等场景下已经体现出CPU确实能已形成瓶颈，而Tungsten就是为了解决这些问题所启动的。Tungsten主要包含三个方面：第一，内存管理与二进制处理；第二，缓存友好的计算；第三，代码生成。

首先着眼第一点，毋庸置疑，JVM本身确实是个非常优秀的平台，但是短板也非常明显，那就是GC，而Java对象的内存开销同样不能忽视。基于这个问题，Spark选择自己管理内存，而所用的工具就是sun.misc.Unsafe。如果大家对细节有兴趣的话，可以关注下BytesToBytesMap这个结构。需要注意它是append-only的，并且key与value都是连续的字节区域。自己管理内存不仅缓解了GC的压力，也显著地降低了内存使用。不过这里必须要提醒下，Unsafe千万不能滥用，否则后果很严重。

其次看下第二点，目前大家看到缓存似乎都只是想到把数据加载内存就完事了，事实上更佳的做法应该是CPU级别的缓存。因此Spark自1.4版本开始就在这个点上发力，其中重中之重当属在 aggregations、joins和shuffle时可以更有效地排序和哈希。Spark引入了UnsafeShuffleManager这个新的ShuffleManager，它的好处是可以直接对二进制数据进行排序，从而不仅减少了内存占用，同时也省去了反序列化这个过程。这里大家可以注意下UnsafeShuffleExternalSorter，可以称得上整个优化的基础。事实上CPU反复从内存读取数据也在一定程度上阻碍了CPU的pipeline 操作。

第三点代码生成（Code Gen）。这点熟悉LLVM的朋友应该能有较好的理解，这个部分其实之前的Spark SQL就已经在使用了，而在近日发布的1.5中得到了更加广泛的应用。需要Code Gen的原因很简单，Code Gen能免去昂贵的虚函数调用，当然也就不存在对Java基本类型装箱之类的操作了。Spark SQL将Code Gen用于了表达式的求值，效果显著。同时，值得一提的是在1.5中，Spark SQL将 Code Gen默认打开了。

Tungsten的部分就先谈到这里，整个项目的完成需要等到1.6版本。不过1.4与1.5已经逐步融入了Tungsten的部分优化，让大家可以及时享受Tungsten带来的各种改进。

Dynamic Resource Allocation

严格来讲，动态资源分配（Dynamic Resource Allocation）这个特性在Spark 1.2的时候就出现了，但是那时只支持在YARN上对资源做动态分配。在Spark 1.5中，Standalone及Mesos也将支持这个特性，个人认为还是有比较大的意义，但是大家仍需去充分了解该特性的使用场景。

在Spark中，动态资源分配的粒度是executor，通过spark.dynamicAllocation.enabled开启，通过spark.dynamicAllocation.schedulerBacklogTimeout及spark.dynamicAllocation.sustainedSchedulerBacklogTimeout两个参数进行时间上的控制。另外说的是，Spark对于YARN及Mesos的支持都得到了显著地增强。

Adaptive Query Plan

适应查询计划（Adaptive Query Plan），一个可能的特性。之所以说是可能的特性，因为这一特性可能要等到Spark 1.6版本或者是之后。首先，陈述几个问题，如何自动确定并行度（Level of Parallelism）；如何自动选择采用broadcast join还是hash join；Spark如何感知数据的行为。目前，Spark需要在执行job前确定job的DAG，即在执行前，由operator到DAG的转换就已经完成了，显然不够灵活。因此，一个更好的方案是允许提交独立的DAG stage，同时收集他们执行结果的一些统计信息。基于这些信息，Spark可以动态决定reduce task的数量，同时也可以动态地选择是采用broadcast还是shuffle。对于Spark SQL来说，应该能在聚合时自动设置reduce task的数量，并且在join时能自动选择策略。主要的思路是在决定reduce task的数量及采用的shuffle策略前，先让map运算然后输出较大数量的partition作为map阶段的结果，接下来Spark会检查map stages输出的partition的大小（或者其它一些状态），然后基于这些信息做出最佳选择。想必大家已经看出来了，这里其实是需要修改DAGScheduler的实现的，因为目前的 DAGScheduler仅仅支持接收一张完整的DAG图，而上述讨论的问题要求DAGScheduler 支持接收map stages，且能收集map stages输出结果的相关信息。Shuffle也需要支持能一次fetch多个map输出的partition，而目前的HashShuffleFetcher 一次性只能获取1个partition。当然，这里还会涉及到其他改动，就不一一列出了。这个特性的重要性在于，Spark会替用户把运行时所需的一些参数及行为确定，从而用户无需操心。还记得Flink那句宣传语吗？“用户对内核唯一需要了解的事就是不需要了解内核”。

概括来讲，Spark护城河其实有两条——其一是先进的技术，另一个则是丰富的生态系统。从图1可以看到，无论是过去这段时间在容器领域无比火爆的kubernete及docker，还是在NoSQL领域的两面锦旗HBase及Cassandra，亦或其它一些如消息队列Kafka，分布式搜索引擎elasticsearch及各机器学习框架都与Spark产生了联系，并且这样的趋势还在快速蔓延中。这意味着，Spark可能出现在大数据处理的各个领域，并给各个领域带来明显提升，从这个角度讲，Spark确实是一个明智的选择。

逆水行舟，看前行中的Spark