最近在看各种博客,学习 spark 源代码。
网上对源代码的分析基本都是基于 0.7, 0.8, 1.0 的代码,而现在的发行版已经是 1.5 了。所以有些代码不大对的上。比如函数 RDD.map()
旧版本是:
def mapU: ClassTag: RDD[U] = new MappedRDD(this, sc.clean(f))
1.5 版本是:
def map[U: ClassTag](f: T => U): RDD[U] = withScope { val cleanF = sc.clean(f) new MapPartitionsRDD[U, T](this, (context, pid, iter) => iter.map(cleanF)) }
而且所有 RDD 的转化方法都调用了 RDDOperationScope.withScope,于是我就很好奇这个函数有什么作用。
它首先是根据堆栈信息(Thread.currentThread.getStackTrace)找出调用者的名字,比如 map, textFile, reduceByKey 等等,然后在 SparkContext 的属性 "spark.rdd.scope" 中新建了一个属性 RDDOperationScope(name: String, parent: RDDOperationScope),用来记录当前的运行 RDD 信息。其中 parent 可以用来追溯到所有 RDD 操作信息,即 RDDOperationScope。
这又啥用?于是我 git blame 了一下,找到了代码的提交哈希值为 fc8b5819,再 git show,于是有了详细的介绍,第一行是
[SPARK-6943] [SPARK-6944] DAG visualization on SparkUI
原来是用来做 DAG 的可视化。再找到这个 SPARK-6943,有非常详细的动机描述:现有的 Spark UI 中只有 stage 的执行情况,而 stage 与用户代码中 rdd 的联系不够直接,如果代码复杂,很难根据 UI 信息了解到代码的执行情况,于是想强化 UI 中的 RDD 可视化功能,所以把所有创建 RDD 的方法包裹起来,使用 RDDOperationScope 记录 RDD 的操作历史和关联,就能达成目标。