Spark中300亿与30亿Parquet表高效全关联及更新优化咨询

一、重构更新逻辑：避免全量Full Join #

直接对300亿+30亿规模的表做Full Join会产生巨量Shuffle数据，建议将逻辑拆分为保留未变更数据+更新匹配数据+插入新增数据三个独立环节，仅处理涉及更新的子集：

// 1. 读取Table2增量数据（按需过滤partition_col，减少处理量）
val table2DF = spark.table("Table2")
  .filter($"partition_col" >= "2024-05-20 00:00:00") // 仅处理目标时间范围的增量
  .select($"key1", $"key2", $"key3", $"col1", $"col2")

// 2. 获取Table1中与Table2匹配的待更新数据
val table1ToUpdateDF = spark.table("Table1")
  .join(table2DF.select($"key1", $"key2", $"key3").distinct(), Seq("key1", "key2", "key3"), "inner")
  .select($"key1", $"key2", $"key3", $"col1", $"col2")

// 3. 生成更新后的数据（用Table2的col1覆盖）
val updatedDF = table1ToUpdateDF.join(table2DF, Seq("key1", "key2", "key3"), "inner")
  .select($"key1", $"key2", $"key3", table2DF("col1").alias("col1"), $"col2")

// 4. 生成新增数据（Table2中不存在于Table1的记录）
val newDF = table2DF.join(table1ToUpdateDF, Seq("key1", "key2", "key3"), "left_anti")
  .select($"key1", $"key2", $"key3", $"col1", $"col2")

// 5. 保留Table1中未被更新的原始数据
val table1UnchangedDF = spark.table("Table1")
  .join(table2DF.select($"key1", $"key2", $"key3").distinct(), Seq("key1", "key2", "key3"), "left_anti")

// 6. 合并最终结果
val finalDF = table1UnchangedDF.union(updatedDF).union(newDF)

二、分区裁剪：减少Table1扫描范围 #

利用col2可由key3推导的特性，先从Table2中提取所有key3，映射到对应的col2分区，仅扫描Table1的相关分区：

// 假设存在key3到col2的映射逻辑（可以是UDF或预定义Map）
def getCol2FromKey3(key3: String): String = {
  // 实现key3到col2的推导逻辑
}

// 提取Table2中所有唯一key3并推导对应的col2
val key3List = table2DF.select($"key3").distinct().collect().map(_.getAs[String]("key3"))
val targetCol2Partitions = key3List.map(getCol2FromKey3).distinct()

// 仅扫描Table1的目标分区
val table1ToUpdateDF = spark.table("Table1")
  .filter($"col2".isin(targetCol2Partitions:_*))
  .join(table2DF.select($"key1", $"key2", $"key3").distinct(), Seq("key1", "key2", "key3"), "inner")
  .select($"key1", $"key2", $"key3", $"col1", $"col2")

三、Join策略与Shuffle参数优化 #

针对大表场景强制使用Shuffle Hash Join，并调整Shuffle参数减少数据传输开销：

// 关闭广播Join，强制使用Shuffle Hash Join
spark.conf.set("spark.sql.autoBroadcastJoinThreshold", -1)
spark.conf.set("spark.sql.join.preferSortMergeJoin", "false")

// 调整Shuffle分区数（建议为总Executor核数的2-3倍：60*4*2=480）
spark.conf.set("spark.sql.shuffle.partitions", "480")

// 开启Shuffle压缩，降低数据量
spark.conf.set("spark.sql.shuffle.compress", "true")
spark.conf.set("spark.sql.shuffle.spill.compress", "true")
spark.conf.set("spark.io.compression.codec", "snappy") // 兼顾压缩比与速度

// 优化Executor内存分配
spark.conf.set("spark.executor.memoryOverhead", "4g") // 32G Executor预留4G堆外内存
spark.conf.set("spark.sql.inMemoryColumnarStorage.compressed", "true") // 开启内存列存储压缩

四、按低基数字段拆分处理 #

利用key3仅1000个唯一值的特性，按key3拆分数据集，实现局部Join，避免全量Shuffle：

// 按key3重分区，将相同key3的数据分配到同一Executor
val table2Repartitioned = table2DF.repartition(1000, $"key3")
val table1Repartitioned = spark.table("Table1").repartition(1000, $"key3")

// 基于key3分区的局部Join，减少跨节点数据传输
val joinDF = table1Repartitioned.join(table2Repartitioned, Seq("key1", "key2", "key3"), "full")

五、进阶方案：使用Lakehouse格式实现高效Merge #

如果允许转换表格式，推荐使用Delta Lake/Iceberg的Merge Into语法，无需全表重写，仅更新/插入目标行：

import io.delta.tables._

// 将Table1转换为Delta表（仅需执行一次）
spark.table("Table1").write.format("delta").saveAsTable("Table1_delta")

// 执行Merge更新
val deltaTable = DeltaTable.forName(spark, "Table1_delta")
deltaTable.as("t1")
  .merge(
    table2DF.as("t2"),
    "t1.key1 = t2.key1 AND t1.key2 = t2.key2 AND t1.key3 = t2.key3"
  )
  .whenMatchedUpdate(set = Map("col1" -> "t2.col1"))
  .whenNotMatchedInsert(values = Map(
    "key1" -> "t2.key1",
    "key2" -> "t2.key2",
    "key3" -> "t2.key3",
    "col1" -> "t2.col1",
    "col2" -> "t2.col2"
  ))
  .execute()

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