PySpark Databricks作业性能排查：阶段耗时与读写/Join分析

代码片段 #

spark.read.table('schema.table_1').createOrReplaceTempView('d1') # 4亿条记录
spark.read.table('schema.table_2').createOrReplaceTempView('d2') # 3亿条记录（注：原代码存在笔误，已修正）

stmt = "select * from d1 inner join d2 on d1.id = d2.id"

(
    spark.sql(stmt).write('delta').mode('overwrite').saveAsTable('schema.table_3') # 结果记录数：8亿条
)

集群配置 #

• Worker节点数：6个
• 单节点规格：32GB内存、4核

DAG执行信息 #

DAG示意图显示：

1. Stage-219 耗时1小时
2. Stage-216与217被跳过
3. Stage-218 耗时40分钟

问题1：Stage-219仅对应写入操作，还是包含SQL语句执行（Join操作）及结果写入目标表的全过程？ #

Stage-219仅对应结果写入目标表的操作。Spark的DAG会将SQL转换类操作（如Join）拆分为独立的计算Stage（如这里的Stage-218），而写入属于Action操作，会单独触发一个Stage完成数据落地。从耗时对比也能佐证：Stage-218的40分钟是Join计算阶段，Stage-219的1小时是后续写入阶段。

问题2：如何区分Join操作与结果写入目标表这两个环节哪个耗时更长？ #

可以通过以下方式明确区分：

• Spark UI Stage统计：直接查看计算类Stage（如Stage-218）和写入类Stage（Stage-219）的总耗时，前者对应Join操作，后者对应写入操作，对比数值即可。
• 中间缓存计时：在SQL执行后、写入前对DataFrame调用cache()并触发count()，分别统计缓存前的计算耗时（Join）和缓存后的写入耗时，通过两次计时结果直接对比。
• 日志时间戳分析：查看Driver和Executor日志，计算阶段会有Shuffle、Join相关日志，写入阶段会有Delta表文件输出日志，通过日志时间戳的差值计算各环节耗时。

问题3：基于DAG信息，Stage-218耗时40分钟，需分析该阶段的性能问题？ #

Stage-218作为Join核心计算阶段，耗时过长的可能原因包括：

1. Shuffle资源不足：Inner Join会触发Shuffle，6个4核Worker总并行度仅24，对于4亿+3亿的大表来说，任务并行度不足；若32GB内存中Shuffle分配比例偏低，会引发磁盘Spill，进一步拖慢处理速度。
2. 数据倾斜：如果id字段存在热点值（如某id对应数百万条记录），会导致部分Task处理数据量远高于其他Task，出现长尾耗时，拉低整个Stage效率。
3. 源表无优化：table_1和table_2未按id分区或建立Z-Order索引，Spark需全表扫描读取数据，增加了数据加载和预处理耗时。
4. 资源配置不合理：单Worker 4核32GB内存的配置中，Executor内存分配可能失衡，引发GC频繁或内存利用率不足；未开启动态资源分配，无法根据任务需求弹性扩容。

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