PySpark两张10万行表连接耗时超半小时，疑与UDF有关

核心结论：UDF确实是性能瓶颈的关键因素 #

Python UDF运行在JVM外部，需要频繁在Python和JVM之间做序列化/反序列化，加上你的UDF里多次调用split字符串操作，在10万行数据量下会累积大量性能开销——这是你和常规场景的核心差异，也是连接变慢的主要嫌疑点。

具体排查与优化步骤 #

1. 用Spark原生函数彻底替换Python UDF #

Spark内置的字符串处理函数（如split、concat_ws）是JVM原生实现，性能远高于Python UDF。把你的UDF逻辑重写为原生函数：

from pyspark.sql.functions import split, concat_ws, col, cast
from pyspark.sql.types import IntegerType

# 替代原UDF的逻辑
df = df.withColumn("split_data", split(col("data"), ","))
# 提取id并转Integer
df = df.withColumn("id", col("split_data")[0].cast(IntegerType()))
# 提取第1到倒数第2个元素，用逗号拼接成f1
df = df.withColumn(
    "f1",
    concat_ws(", ", col("split_data").slice(1, col("split_data").size() - 1))
)
# 提取最后一个元素转Integer作为f2
df = df.withColumn("f2", col("split_data")[-1].cast(IntegerType()))
# 清理临时列
df = df.drop("data", "split_data")

替换后重新生成b、c表，再测试连接性能——这是最可能解决问题的步骤。

2. 验证数据分布与统计信息 #

虽然你说不存在数据倾斜，但仍需确认：

• 执行b.groupBy("id").count().orderBy(col("count").desc()).show(20)，检查是否有单个id对应远超平均的行数（比如某id对应几万行，这会导致Shuffle时单个任务过载）。
• 手动收集表的统计信息，帮助Spark优化器生成更优执行计划：

  # 缓存表并触发计算，生成统计信息
  b.cache().count()
  c.cache().count()
  # 或者用SQL命令收集列级统计
  spark.sql("ANALYZE TABLE b COMPUTE STATISTICS FOR COLUMNS id")
  spark.sql("ANALYZE TABLE c COMPUTE STATISTICS FOR COLUMNS id")
  

3. 检查Spark UI的Shuffle与任务细节 #

打开Spark UI的SQL和Jobs页面，重点看：

• 连接阶段的Shuffle Read/Write数据量是否异常大。
• 单个任务的执行时间是否差异悬殊（即使整体无倾斜，部分任务可能因数据序列化/反序列化变慢）。
• 确认spark.sql.adaptive.enabled=true是否生效（SQL页面会显示Adaptive Execution相关日志）。

4. 物化UDF生成的数据 #

UDF生成的DataFrame是逻辑执行计划，每次操作可能会重新计算。可以先将生成的b、c表写入Parquet文件，再读出来做连接，避免重复计算UDF逻辑：

# 写入Parquet
b.write.mode("overwrite").parquet("/path/to/b_table")
c.write.mode("overwrite").parquet("/path/to/c_table")
# 重新读取
b = spark.read.parquet("/path/to/b_table")
c = spark.read.parquet("/path/to/c_table")
# 再执行连接
a = b.join(c, b.id == c.id, "inner").drop(c.id)

5. 排查数据生成阶段的性能 #

先单独测试UDF生成10万行表的耗时，如果生成过程本身就需要十几分钟，那连接阶段的慢是连锁反应——替换为原生函数后，生成速度应该会大幅提升。

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