转载

逆水行舟，看前行中的Spark

　　过去两年，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