为何Spark为这段PySpark代码生成4个Job和6个Stage？

你提到代码里只有2个show()动作，却生成了4个Job；2个groupBy对应4个Stage，实际却有6个，以下是具体原因：

代码回顾 #

sc = SparkContext()
glueContext = GlueContext(sc)
spark = glueContext.spark_session
job = Job(glueContext)

_schema = "InvoiceID string, Branch string, City string, CustomerType string, Gender string, \
           ProductLine string, UnitPrice double, Quantity integer, Tax_5 double, Total double, \
           Date date, Time timestamp, Payment string, CustomeID string"

df = spark.read.format("csv").schema(_schema) \
        .option("header", True) \
        .option("sep", ",") \
        .load("s3://*******************/test/sales/")

# df = df.repartition("City", "Branch", "ProductLine")

df = df.where("CustomerType = 'Member'")

df1 = df.groupBy("City", "Branch", "ProductLine").agg(
    sum(col("Quantity")).alias("total_quantity"),
    avg(col("UnitPrice")).alias("avg_unit_price")
)

df1 = df1.where("Branch in ('A','B')").withColumn("avg_unit_price", col("avg_unit_price")-100)

df1.explain(mode="formatted")
df1.show()

df2 = df.groupBy("City", "Branch").agg(
    sum(col("Quantity")).alias("total_quantity"),
    avg(col("UnitPrice")).alias("avg_unit_price")
)

df2 = df2.where("Branch in ('A')").withColumn("avg_unit_price", col("avg_unit_price")-10)
df2.explain(mode="formatted")
df2.show()

Job数量为什么是4？ #

你忽略了代码中的两个explain(mode="formatted")调用。普通explain()仅输出逻辑计划，不会触发计算，但formatted模式需要获取数据的统计信息（比如分区数、数据规模）来生成结构化的执行计划，这会触发Spark执行一个Job收集元数据。所以：

• 2个show() → 2个Job
• 2个explain(mode="formatted") → 2个Job
• 总计4个Job

Stage数量为什么是6？ #

Stage的划分核心是Shuffle操作（宽依赖），每个groupBy聚合都会触发Shuffle，将执行计划拆分为两个Stage：

1. Shuffle前Stage：读取CSV、执行CustomerType='Member'过滤、完成map端聚合，输出Shuffle文件；
2. Shuffle后Stage：读取Shuffle文件、完成reduce端聚合、执行后续过滤和列变换。

具体到你的代码：

• 针对df1的两个Job（explain和show），每个对应2个独立Stage，共4个；
• 针对df2的两个Job，由于df2基于同一过滤后的df，Spark复用了之前已执行的Shuffle前Stage，每个Job仅需1个新的Shuffle后Stage，共2个；
• 总计4+2=6个Stage

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