如何合并Spark中summary生成DF与自定义指标DF为指定格式？

核心思路 #

把自定义指标的宽表转换成和summary()输出一致的长表结构，再和原summary结果合并、去重、排序，就能得到目标格式。

具体步骤（Python版） #

假设原summary生成的DataFrame叫summary_df，自定义指标的DataFrame叫custom_df。

1. 转换自定义指标为长表结构
   把宽表形式的自定义指标拆成和summary匹配的行结构，同时对应好每个指标的列值：

from pyspark.sql import functions as F

# 拆解自定义指标，生成符合summary格式的行
custom_metrics = custom_df.select(
    F.explode(F.array(
        # 去重指标行
        F.struct(
            F.lit("dstnct").alias("summary"),
            F.col("distinct_col1").alias("col1"),
            F.col("distinct_col2").alias("col2"),
            F.col("distinct_col3").alias("col3")
        ),
        # 完整性指标行
        F.struct(
            F.lit("complt").alias("summary"),
            F.col("complete_col1").alias("col1"),
            F.col("complete_col2").alias("col2"),
            F.col("complete_col3").alias("col3")
        ),
        # col3的最小值行
        F.struct(
            F.lit("min").alias("summary"),
            F.lit(None).alias("col1"),
            F.lit(None).alias("col2"),
            F.col("min_col3").alias("col3")
        ),
        # col3的最大值行
        F.struct(
            F.lit("max").alias("summary"),
            F.lit(None).alias("col1"),
            F.lit(None).alias("col2"),
            F.col("max_col3").alias("col3")
        )
    )).alias("metrics")
).select("metrics.*")

2. 处理原summary结果
   给原summary补充col3列（默认空值），同时过滤掉原有的min、max行（因为自定义指标里有col3的对应值）：

# 补充col3列
summary_with_col3 = summary_df.withColumn("col3", F.lit(None))
# 过滤原summary的min、max行
filtered_summary = summary_with_col3.filter(~F.col("summary").isin("min", "max"))

3. 合并并排序
   将处理后的两个DataFrame合并，再按目标顺序排序：

# 合并两个DataFrame
combined_df = filtered_summary.unionByName(custom_metrics, allowMissingColumns=True)

# 按指定顺序排序，保证行的顺序和目标一致
target_order = ["count", "dstnct", "complt", "mean", "stddev", "min", "25%", "50%", "75%", "max"]
final_df = combined_df.orderBy(
    F.expr(f"array_position(array({','.join([f'\'{x}\'' for x in target_order])}), summary)")
)

4. 查看结果
   执行final_df.show(truncate=False)就能得到和目标格式一致的输出。

注意事项 #

• 如果col3是日期类型，要确保custom_df里的min_col3、max_col3已经转成日期类型，避免类型不兼容。
• 如果用Scala实现，逻辑完全一致，只是语法略有不同（比如lit的调用、数组构造方式）。

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