Spark WordCount任务DAG解读及阶段划分规则咨询

首先附上你的WordCount代码：

sc = SparkContext("local","PySpark Word Count Exmaple")
print("0:",type(sc))
print("0:",sc)

# read data from text file and split each line into words
rdd = sc.textFile("file:///home/prashant/hello.txt")
print("1:",type(rdd))
print("2:",rdd)
words=rdd.flatMap(lambda line: line.split(" "))
print("3:",type(words))
print("4:",words)

# count the occurrence of each word
wordmap = words.map(lambda word: (word, 1))
print("5:",type(wordmap))
print("6:",wordmap)
wordCounts=wordmap.reduceByKey(lambda a,b:a +b)
print("7:",type(wordmap))
print("8:",wordmap)

# save the counts to output
wordCounts.saveAsTextFile("file:///home/prashant/spark_output1")

问题1：如何解读该DAG并理解其含义？ #

Spark的DAG（有向无环图）是任务执行逻辑的可视化呈现，每个节点对应一个RDD转换操作，节点间的边代表RDD的依赖关系，Stage（阶段）则是Spark根据依赖类型划分的可并行执行任务组：

• Stage 0：包含从textFile读文件、flatMap拆分单词、map生成键值对，再到reduceByKey的本地聚合阶段。这些操作都是窄依赖——每个父RDD分区仅对应一个子RDD分区，数据无需跨节点传输，可在同一个分区内流水线执行。reduceByKey在这里会先对每个分区内的相同key做本地聚合，减少后续shuffle的数据量。
• Stage 1：以shuffle操作为起点，承接Stage0输出的shuffle数据，完成reduceByKey的全局聚合，最后执行saveAsTextFile输出结果。这个阶段处理跨分区的相同key聚合，需要依赖shuffle传输过来的其他分区数据。

简言之，DAG清晰展示了任务从数据读取、转换到最终输出的全流程，Stage的划分则体现了Spark的执行优化思路：把无需跨节点传输的操作打包在一起，降低网络开销。

问题2：reduceByKey属于宽转换，且是最后执行的转换操作，为何它出现在Stage 0而非预期的Stage 1？ #

这是因为reduceByKey并非单一操作，它包含两个逻辑步骤：

1. map-side combine（本地聚合）：在每个分区内部，对相同key的数值先做一次聚合，这一步属于窄依赖——仅用到当前分区的数据，无需和其他分区交互，因此会被归到前面的Stage（Stage0），和flatMap、map等操作一同执行。
2. reduce-side aggregate（全局聚合）：需要将所有分区中相同key的数据通过shuffle拉取到一起，再做最终聚合，这一步属于宽依赖，会触发Stage划分，进入Stage1。

你在DAG的Stage0里看到的reduceByKey，其实是它的本地聚合部分，而真正需要shuffle的全局聚合在Stage1中执行——这是Spark对reduceByKey的优化设计，目的是减少shuffle的数据量。

问题3：Spark是如何将转换函数划分为不同阶段的？ #

Spark划分Stage的核心依据是依赖类型，具体规则如下：

• 以**宽依赖（shuffle依赖）**作为Stage的拆分边界：当遇到需要跨分区传输数据的操作（比如reduceByKey、groupByKey、join等），就会在这个操作前拆分出一个新的Stage。
• 从任务的**最后一个动作（Action）**往前回溯：比如你的任务里的saveAsTextFile是动作，从它开始往前查找，遇到第一个宽依赖就拆分Stage，前面的所有窄依赖操作归为一个Stage，宽依赖之后的操作归为下一个Stage。
• 每个Stage包含一组连续的窄依赖操作：这些操作可在同一个分区内流水线执行，无需跨节点数据传输，能最大化执行效率。
• 特殊优化：像reduceByKey、aggregateByKey这类自带本地聚合的操作，会把本地聚合部分归到前一个Stage，shuffle后的全局聚合归到下一个Stage，避免把整个操作都放到需要shuffle的Stage里。

内容的提问来源于stack exchange，提问作者nayak0765