Spark SQL中整个代码生成的"阶段"如何与Spark Core的阶段相关？

Question

我一直在探索Spark SQL中的全阶段代码生成优化(又称全阶段codegen),并且想知道"while-stage"中的"阶段"是多少来自Spark Core的一个阶段(Spark工作)的含义？

Spark SQL和Spark Core中的整个阶段代码阶段之间是否存在任何技术关系？或者它们是否更广泛地用于指计算中的"阶段"？

Answer 1

"stage"的术语(和概念)在RDD执行,SQL/DataFrame执行和"Wholestage Codegen"中是相同的.

阶段是指从读取(从外部源或先前的shuffle)到后续写入(到下一个shuffle或最终输出位置,如文件系统,数据库等)的所有窄(映射)操作.

使用Wholestage Codegen,在可能的情况下,每个物理运算符生成一些代码并将它们"融合在一起"(基于像Lego砖这样的常见模式)来制作一个可编译的大型Java函数 - 称之为"f"

然后通过(大致)获取RDD[InternalRow]所需的字段/列并执行a来完成执行rdd.mapPartitions(f)

查看它的另一种方法是在SQL UI选项卡中:当"完整"的Wholestage Codegen实现时,蓝色外部codegen框覆盖除Exchange框之外的所有内容(物理shuffle)

Answer 2

这些概念非常相似,但并不总是相同.

在Spark Core Stage中,对应于shuffle边界内的一组运算符.

下面的表达式中的explain()函数已经扩展用于整个阶段代码生成.在explain输出中,当运算符周围有星(*)时,启用整阶代码生成.在下面的例子中,Range,Filter和两个Aggreg都在运行全阶段代码.但是,Exchange不会实现整阶代码生成,因为它通过网络发送数据.

spark.range(1000).filter("id > 100").selectExpr("sum(id)").explain()

== Physical Plan ==
*Aggregate(functions=[sum(id#201L)])
+- Exchange SinglePartition, None
   +- *Aggregate(functions=[sum(id#201L)])
      +- *Filter (id#201L > 100)
         +- *Range 0, 1, 3, 1000, [id#201L]

在Full-Stage-Codegen的情况下,CollapseCodegenStages物理准备规则用于查找支持codegen的计划并将它们作为WholeStageCodegen折叠在一起.

请仔细阅读以下链接以获得更好的主意.

https://jaceklaskowski.gitbooks.io/mastering-apache-spark/spark-sql-whole-stage-codegen.html

https://jaceklaskowski.gitbooks.io/mastering-apache-spark/spark-sql-CollapseCodegenStages.html

https://databricks.com/blog/2016/05/23/apache-spark-as-a-compiler-joining-a-billion-rows-per-second-on-a-laptop.html