PySpark中DataFrame.count()多任务及执行计划疑问

核心原因：Schema推断带来的额外Stage #

你看到的3个Stage，大概率是Spark自动推断CSV Schema时触发了一个隐式前置Stage：

• Stage 0：Spark启动单个任务扫描CSV文件的部分数据（通常是首行），用来推断列名和数据类型。这个步骤是自动执行的，只要你没提前指定Schema就会触发。
• Stage 1：扫描所有CSV文件分区，解析数据并对每个分区计算局部行数（Map阶段，任务数等于CSV的分区数）。
• Stage 2：通过Shuffle汇总各分区的局部行数，计算全局总数（Reduce阶段，任务数由spark.sql.shuffle.partitions配置决定，默认200，数据量小时可能合并为1个）。

如何简化执行计划 #

如果不想生成这个额外Stage，直接提前指定CSV的Schema即可，示例代码：

from pyspark.sql.types import StructType, StructField, StringType, IntegerType

# 定义Schema
schema = StructType([
    StructField("col1", StringType(), nullable=True),
    StructField("col2", IntegerType(), nullable=True)
])

# 读取CSV时指定Schema
df = spark.read.csv("your_file_path.csv", schema=schema, header=True)
df.count()

这样Spark无需额外扫描推断Schema，执行计划会简化为2个Stage：扫描解析+局部计数为一个Stage，全局聚合为另一个Stage。

Spark Stage划分逻辑补充 #

Spark的Stage由**宽依赖（Shuffle操作）**分割：

• 每个Shuffle操作（如聚合、Join）会将DAG拆分为前后两个Stage。
• 自动Schema推断的扫描是独立的窄依赖Stage，仅执行元数据扫描任务，不需要Shuffle。

你提供的执行计划图 #

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