Bug剖析篇-“Facebook 60TB+级的Apache Spark应用案例”

  Facebook 60TB+级的Apache Spark应用案例 里大体有两方面的PR,一个是Bug Fix,一个是性能优化。这篇文章会对所有提及的Bug Issue进行一次解释和说明。也请期待下一篇。

  前言

  Facebook 60TB+级的Apache Spark应用案例 ,本来上周就准备看的,而且要求自己不能手机看,要在电脑上细细的看。然而终究是各种忙拖到了昨天晚上。

  文章体现的工作,我觉得更像是一次挑战赛,Facebook团队通过层层加码,最终将单个Spark Batch实例跑到了60T+ 的数据,这是一个了不起的成就,最最重要的是,他们完成这项挑战赛后给社区带来了三个好处:

  在如此规模下,发现了一些Spark团队以前很难发现的Bug

  提交了大量的bug fix 和 new features,而且我们可以在Spark 1.6.2 /Spark 2.0 里享受到其中的成果

  在如此规模下,我们也知道我们最可能遇到的一些问题。大体是OOM和Driver的限制。

  说实在的,我觉得这篇文章,可以算是一篇工程论文了。而且只用了三个人力,不知道一共花了多久。

  值得注意的是,大部分Bug都是和OOM相关的,这也是Spark的一个痛点,所以这次提交的PR质量非常高。

  Bug 剖析

  Make PipedRDD robust to fetch failure SPARK-13793

  这个Issue 还是比较明显的。PipedRDD 在Task内部启动一个新的Java进程(假设我们叫做ChildProcessor)获取数据。这里就会涉及到三个点:

  启动一个线程往 ChildProcessor 写数据 (stdin writer)

  启动一个线程监控ChildProcessor的错误输出 (stderr reader)

  获取ChildProcessor输入流,返回一个迭代器(Iterator)

  既然都是读取数据流,如果数据流因为某种异常原因关闭,那必然会抛出错误。所以我们需要记录这个异常,对于1,2 两个我们只要catch住异常,然后将异常记录下来方便后续重新抛出。 那么什么时候抛出呢?迭代器有经典的hasNext/next方法,每次hasNext时,我们都检查下是否有Exception(来自1,2的),如果有就抛出了。既然已经异常了,我们就应该不需要继续读取这个分区的数据了。否则数据集很大的情况下,还要运行很长时间才能运行完。

  在hasNext 为false的情况下,有两类情况,一类是真的没有数据了,一类是有异常了,比如有节点挂了,所以需要检测下ChildProcessor的exitStatus状态。如果不正常,就直接抛出异常,进行重试。

  对于1,2两点,原来都是没有的,是这次Facebook团队加上去的。

  Configurable max number of fetch failures SPARK-13369

  截止到我这篇文章发出,这个Issue 并没有被接收。

  我们知道,Shuffle 发生时,一般会发生有两个Stage 产生,一个ShuffleMapStage (我们取名为 MapStage),他会写入数据到文件中,接着下一个Stage (我们取名为ReduceStage) 就会去读取对应的数据。 很多情况下,ReduceStage 去读取数据MapStage 的数据会失败,可能的原因比如有节点重启导致MapStage产生的数据有丢失,此外还有GC超时等。这个时候Spark 就会重跑这两个Stage,如果连续四次都发生这个问题,那么就会将整个Job给标记为失败。 现阶段(包括在刚发布的2.0),这个数值是固定的,并不能够设置。

[email protected] 给出的质疑是,如果发生节点失败导致Stage 重新被Resubmit ,Resubmit后理论上不会再尝试原来失败的节点,如果连续四次都无法找到正常的阶段运行这些任务,那么应该是有Bug,简单增加重试次数虽然也有意义,但是治标不治本。

  我个人认为在集群规模较大,任务较重的过程中,出现一个或者一批Node 挂掉啥的是很正常的,如果仅仅是因为某个Shuffle 导致整个Job失败,对于那种大而耗时的任务显然是不能接受的。个人认为应该讲这个决定权交给用户,也就是允许用户配置尝试次数。

  Unresponsive driver SPARK-13279

  这个Bug已经在1.6.1, 2.0.0 中修复。 这个场景比较特殊,因为Facebook产生了高达200k的task数,原来给 pendingTasksForExecutor:HashMap[String, ArrayBuffer[Int]] 添加新的task 的时候,都会根据Executor名获取到已经存在的列表,然后判断该列表是否已经包含了新Task,这个操作的时间复杂度是O(N^2)。在Task数比较小的情况下没啥问题,但是一旦task数达到了200k,基本就要五分钟,给人的感觉就是Driver没啥反应了。

  而且在实际运行任务的过程中,会通过一个特殊的dequeueTaskFromList结构来排除掉已经运行的任务,所以我们其实在addPendingTask 过程中不需要做这个检测。