PySpark执行DataFrame分组统计任务过慢该如何排查优化？

• 检查DataFrame字段解析逻辑是否缺失。使用spark.read.text()读取txt文件，默认只会生成单一value列存储每行完整文本，你提到的city_name/country_name等字段并未实际解析生成。如果后续分组代码未报错，大概率是省略了字段解析逻辑，若字段解析是用map/udf这类逐行处理的算子实现，没有做向量化优化，20万行数据即使在本地也会出现明显耗时。
• 检查当前运行模式的资源配置。master("local[*]")是本地多线程模式，本身没有独立的executor节点，所有计算都在driver进程的多线程中运行。如果本地CPU核心数少、内存不足，或者Spark默认给driver分配的内存过小（默认1G），都会拖慢计算速度。
• 检查任务是否存在重复计算。两次show()都是行动算子，会触发两次完整的读取、解析、分组、排序流程，相当于重复跑了两遍全链路任务，耗时直接翻倍。

• 完善结构化读取逻辑，避免逐行解析。如果txt是分隔符格式（比如csv、tsv），直接用spark.read.csv()指定分隔符、表头、字段类型，直接生成带对应字段的DataFrame，比读text后自行解析性能高几个量级：

# 示例：csv格式，分隔符为\t，第一行不是表头
df = spark.read.csv(
    filepath,
    sep="\t",
    schema="timestamp long, user_hash string, browser_name string, os_name string, city_name string, country_name string"
)

• 添加缓存避免重复计算。如果要多次对同一个DataFrame做计算，先对解析后的df做缓存，只跑一次读取解析流程：

df = df.cache()
# 第一次action会触发缓存
df.groupBy("city_name").count().orderBy(desc("count")).show(5)
# 第二次action直接读缓存数据
df.groupBy("country_name").count().orderBy(desc("count")).show(5)
# 用完后释放缓存
df.unpersist()

• 优化Spark Session配置，给本地运行分配足够资源：

spark = SparkSession.builder \
    .master("local[*]") \
    .appName("test") \
    .config("spark.driver.memory", "4g") # 根据本地内存调整，建议至少2g
    .config("spark.sql.shuffle.partitions", "10") # 本地运行不需要默认200个shuffle分区，减少调度开销
    .getOrCreate()

• 替换排序逻辑为limit下推优化，避免全量排序。直接在排序后取5条，Spark会自动优化为每个分区先取Top5再合并，无需全量排序：

df.groupBy("city_name").count().orderBy(desc("count")).limit(5).show()

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