如何在PySpark中高效连接7张Iceberg表（含大、中、小表）

我在Databricks中通过以下代码读取7张Iceberg表为DataFrame：

df1 = spark.read.format('iceberg').load(path1) # 行数：2.5亿
df2 = spark.read.format('iceberg').load(path2) # 行数：2.5亿
df3 = spark.read.format('iceberg').load(path3) # 行数：200万
df4 = spark.read.format('iceberg').load(path4) # 行数：7
df5 = spark.read.format('iceberg').load(path5) # 行数：11
df6 = spark.read.format('iceberg').load(path6) # 行数：27
df7 = spark.read.format('iceberg').load(path7) # 行数：31

需要以df1为左表，与其余6张DataFrame执行左连接。当前代码执行耗时超1小时，任务多且部分任务运行时间过长，代码如下：

broadcast_df4 = broadcast(df4)
broadcast_df5 = broadcast(df5)
broadcast_df6 = broadcast(df6)
broadcast_df7 = broadcast(df7)

joined_df = df1.join(broadcast_df4, id4, "left")
               .select(df1.col1, broadcast_df4.col2, broadcast_df4.col3)
               .join(broadcast_df5, roll, "left")
               .select(broadcast_df5.name)
               # 其余小表的连接逻辑类似
               .join(df2, df1.acc == df2.key, "left").select(df2.col5, df2.col6...)
               .join(df3, df1.c_id == df3.a_id, "left").select(df3.col9, df3.col10...)
joined_df.count()

已知广播小表可优化连接，但不知道如何高效连接大表（df2）和中表（df3），请问有没有更优方案？使用Spark SQL是否能提升效率？

1. 调整连接逻辑，避免不必要的数据丢弃 #

• 不要在每次join后立即调用select丢弃df1的关联键，否则后续连接大表/中表时，Spark无法基于原表的分区或索引做优化，还会导致数据重复处理。应保留所有需要的关联键和最终输出字段，最后统一执行select。
• 优先完成所有小表的广播连接，再处理大表与中表的连接，减少大表数据的流转次数。

2. 大表-大表连接（df1与df2）优化 #

• 利用Iceberg的分区与二级索引：确认两张大表是否在连接键（df1.acc、df2.key）上设置了分区或Iceberg二级索引，Spark会自动利用这些结构裁剪扫描的数据量，大幅减少IO开销。
• 调优Shuffle Join参数：
  • 修改spark.sql.shuffle.partitions默认值（200），根据数据量调整到合理范围（比如2000-5000），确保每个Shuffle分区的数据量控制在100MB左右，避免单个分区过大导致任务超时。
  • 确保开启spark.sql.adaptive.enabled=true（Databricks默认开启），让Spark自动根据数据量调整分区数和执行计划。
• 预过滤字段：读取表时直接筛选出需要的字段，减少数据传输和处理量：

  df1 = spark.read.format('iceberg').load(path1).select("acc", "col1", "c_id", "id4", "roll")
  df2 = spark.read.format('iceberg').load(path2).select("key", "col5", "col6")
  

3. 中表（df3）连接优化 #

• 尝试广播中表：200万行的表如果单条数据较小（总数据量≤200MB），可以调整spark.sql.autoBroadcastJoinThreshold参数（比如设置为209715200即200MB），或者手动调用broadcast(df3)，避免Shuffle操作带来的开销。
• 若中表不适合广播，同样利用Iceberg的分区/索引裁剪数据，配合自适应执行计划优化Shuffle过程。

4. Spark SQL与DataFrame API的效率对比 #

两者在执行效率上没有本质差异，因为最终都会被Spark Catalyst优化为相同的物理执行计划。但SQL写法在多表连接场景下更简洁易读，维护成本更低：

-- 先将DataFrame注册为临时视图
df1.createOrReplaceTempView("df1")
df2.createOrReplaceTempView("df2")
df3.createOrReplaceTempView("df3")
df4.createOrReplaceTempView("df4")
df5.createOrReplaceTempView("df5")

SELECT 
  df1.col1,
  df4.col2, df4.col3,
  df5.name,
  df2.col5, df2.col6,
  df3.col9, df3.col10
FROM df1
LEFT JOIN df4 ON df1.id4 = df4.id4
LEFT JOIN df5 ON df1.roll = df5.roll
-- 补充其余小表的连接逻辑
LEFT JOIN df2 ON df1.acc = df2.key
LEFT JOIN df3 ON df1.c_id = df3.a_id

5. 通用排查与调优 #

• 查看执行计划：执行joined_df.explain()查看物理执行计划，确认小表是否被广播，大表连接是否用到了分区裁剪，是否存在不必要的Shuffle操作。
• 集群资源调优：确保Databricks集群有足够的CPU和内存资源，避免因资源不足导致任务排队；选择内存优化型实例，提升大数据处理的效率。

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