如何将分组后的DataFrame保存至不同Delta Live Tables

针对你的需求，以下是无需依赖Pandas UDF、直接基于PySpark原生API实现分组写入不同Delta表的方案，同时解决键倾斜导致的效率问题：

核心方案：使用mapGroups实现分组自定义写入 #

mapGroups可以直接对groupBy后的分组进行自定义处理，每个分组仅需处理一次，避免多次filter重复扫描原数据的问题。

代码实现 #

from pyspark.sql import SparkSession
from pyspark.sql.types import StructType

def write_group_to_delta(key, iterator):
    # 获取当前活跃的SparkSession（Executor端可用）
    spark = SparkSession.getActiveSession()
    if not spark:
        spark = SparkSession.builder.getOrCreate()
    
    # 从迭代器构建分组DataFrame，复用原DataFrame的Schema
    group_schema = my_df.schema
    group_df = spark.createDataFrame(iterator, schema=group_schema)
    
    # 写入对应名称的Delta Live Table
    target_table = key[0]  # groupBy的key为元组，取第一个元素作为表名
    group_df.write.format("delta") \
        .mode("append")  # 根据需求选择append/overwrite等模式
        .saveAsTable(target_table)
    
    # 返回空迭代器（mapGroups要求返回数据，此处仅做写入无需返回）
    return iter([])

# 触发执行：mapGroups是懒执行，需通过count()等动作触发
my_df.groupBy("table_name") \
    .mapGroups(write_group_to_delta, schema=StructType([])) \
    .count()

键倾斜优化：分组加盐拆分 #

如果存在严重的键倾斜（某几个分组数据量极大），可以通过加盐拆分大分组，将单个大任务拆分为多个小任务，避免Executor内存溢出或任务超时：

from pyspark.sql.functions import col, concat, lit, floor, rand

# 对倾斜的表名添加随机盐值，拆分为多个子分组
salted_df = my_df.withColumn(
    "salted_table",
    # 假设已知倾斜表名，或通过数据统计自动识别
    when(
        col("table_name").isin("large_table_01", "large_table_02"),
        concat(col("table_name"), lit("_"), floor(rand() * 10).cast("string"))
    ).otherwise(col("table_name"))
)

def write_salted_group(key, iterator):
    spark = SparkSession.getActiveSession()
    group_schema = my_df.schema
    group_df = spark.createDataFrame(iterator, schema=group_schema)
    
    # 还原原始表名
    original_table = key[0].split("_")[0] if "_" in key[0] else key[0]
    group_df.write.format("delta") \
        .mode("append") \
        .saveAsTable(original_table)
    
    return iter([])

# 基于加盐后的分组执行写入
salted_df.groupBy("salted_table") \
    .mapGroups(write_salted_group, schema=StructType([])) \
    .count()

方案优势 #

1. 效率提升：仅需一次Shuffle将数据按分组聚合，避免filter方案多次全表扫描的开销。
2. 原生Spark支持：无需依赖Pandas，直接在Executor端处理分组写入，适配Spark分布式架构。
3. 倾斜处理：通过加盐拆分大分组，解决单个分组数据量过大导致的性能问题。

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