如何在Databricks中使用Spark高效合并17张异构增量表？

核心思路 #

针对17张仅id和date为公共列的源表，最优方式是基于(id, date)联合键的动态全外连接（Full Outer Join），结合Delta Lake实现增量处理与存储，确保大数据量下的性能与可维护性。

具体实现步骤 #

1. 统一读取增量数据 #

每日仅读取各源表中date等于当日的增量数据（若源表未按date分区，需通过过滤条件精准筛选）：

from pyspark.sql import SparkSession

spark = SparkSession.builder.getOrCreate()

# 定义源表清单，格式：(表标识, 读取路径/元数据表名)
source_tables = [
    ("source1", "/path/to/source1"),
    ("source2", "/path/to/source2"),
    # ... 补充剩余15张表的信息
]

# 获取当日日期（可通过Spark函数自动生成，示例为固定值）
current_date = "2023-02-01"
daily_dfs = []
for _, path in source_tables:
    # 按源表实际格式读取，这里以Delta为例
    df = spark.read.format("delta").load(path)
    # 筛选当日增量数据
    df = df.filter(df.date == current_date)
    daily_dfs.append(df)

2. 动态全外连接合并 #

利用reduce函数批量执行全外连接，自动合并所有唯一列：

from functools import reduce
from pyspark.sql import DataFrame

def full_join(df1: DataFrame, df2: DataFrame) -> DataFrame:
    return df1.join(df2, on=["id", "date"], how="full_outer")

# 合并所有当日增量DataFrame
merged_daily_df = reduce(full_join, daily_dfs)

3. 增量写入目标表 #

目标表采用Delta Lake格式，通过MERGE操作实现幂等性增量更新，避免重复数据：

target_table_path = "/path/to/target_table"

# 临时注册当日合并结果
merged_daily_df.createOrReplaceTempView("daily_increment")

# 执行MERGE：匹配(id, date)则更新全量列，不匹配则插入
spark.sql(f"""
    MERGE INTO delta.`{target_table_path}` t
    USING daily_increment s
    ON t.id = s.id AND t.date = s.date
    WHEN MATCHED THEN UPDATE SET *
    WHEN NOT MATCHED THEN INSERT *
""")

关键性能优化点 #

• Delta Lake自动优化：开启目标表的自动优化与压缩，减少小文件、优化数据布局：

  ALTER TABLE delta.`{target_table_path}` SET TBLPROPERTIES (
    'delta.autoOptimize.optimizeWrite' = 'true',
    'delta.autoOptimize.autoCompact' = 'true'
  )
  

• 分区裁剪：确保源表和目标表均按date分区，每日仅扫描当日分区，大幅减少数据处理量。
• 广播小表：对每日增量数据量极小的表（如<1GB），使用broadcast函数广播，避免Shuffle开销：

  from pyspark.sql.functions import broadcast
  
  def optimized_full_join(df1, df2):
      if df2.count() < 1000000:  # 根据实际场景调整阈值
          return df1.join(broadcast(df2), on=["id", "date"], how="full_outer")
      return df1.join(df2, on=["id", "date"], how="full_outer")
  

• 资源调优：在Databricks集群中配置合理的Executor资源（如16GB内存+6核/Executor），开启动态资源分配，匹配50GB日处理量的需求。
• 列名校验：提前检查源表列名，避免除id/date外的列名重复，防止join后出现col1/col1_1这类冗余列。

注意事项 #

• 若源表date格式不统一（如示例中的01.02.2023），需先转换为Spark标准日期格式（yyyy-MM-dd），避免过滤或join逻辑出错。
• 每日处理前可校验各源表的增量数据量，若无数据则跳过该表的读取与join，节省计算资源。

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