Spark Scala：如何为DataFrame添加Schema不同的行？附示例

首先，从你给出的DataFrame结构来看，核心需求是在原有行的基础上补充结构（Schema）不同的行——这类需求的关键是先确保最终合并后的DataFrame有统一的Schema（否则无法完成合并），再根据具体目标构造新增行。下面我结合两种最常用的DataFrame工具（Pandas和PySpark），给出两种常见场景的实现方法：

场景1：添加带汇总标记的兼容行 #

如果你的目标是为每个id添加一行汇总信息（比如整合该id下的所有v1/v2/v3值），这类行的Schema可以和原DataFrame兼容，仅新增一个标记列区分行类型：

原DataFrame结构回顾 #

Pandas 代码实现 #

import pandas as pd

# 构造示例DataFrame（已有可跳过）
data = [
    (1, 'sc1', 'ok', None, None),
    (1, 'sc2', 'no', None, None),
    (1, 'sc3', 'yes', None, None),
    (1, 'sc4', None, '20180318', None),
    (1, 'sc5', None, None, '["5","2","9"]'),
    (1, 'sc6', None, '20180317', None),
    (1, 'sc7', 'ok++', None, None),
    (2, 'sc3', 'yes++', None, None),
    (2, 'sc2', 'no--', None, None),
    (2, 'sc7', 'ok--', None, None),
    (2, 'sc4', None, '20180315', None),
    (3, 'sc1', 'no', None, None),
    (3, 'sc6', None, '20180313', None)
]
df = pd.DataFrame(data, columns=['id', 'k', 'v1', 'v2', 'v3'])

# 生成每个id的汇总行
summary_rows = []
for id_val in df['id'].unique():
    subset = df[df['id'] == id_val]
    # 整合非空值：v1用逗号拼接，v2取最早日期，v3用分号拼接
    v1_combined = ', '.join(subset['v1'].dropna().tolist()) or None
    v2_earliest = min(subset['v2'].dropna().tolist()) if not subset['v2'].dropna().empty else None
    v3_combined = '; '.join(subset['v3'].dropna().tolist()) or None
    
    summary_rows.append({
        'id': id_val,
        'k': 'summary',  # 标记为汇总行
        'v1': v1_combined,
        'v2': v2_earliest,
        'v3': v3_combined,
        'row_type': 'summary'  # 新增列区分行类型
    })

# 合并原DataFrame和汇总行
summary_df = pd.DataFrame(summary_rows)
final_df = pd.concat([df, summary_df], ignore_index=True)

# 查看id=1的汇总行
print(final_df[final_df['id'] == 1].tail(1))

PySpark 代码实现 #

from pyspark.sql import SparkSession
from pyspark.sql import functions as F

spark = SparkSession.builder.appName("AddSummaryRows").getOrCreate()

# 构造示例DataFrame
data = [
    (1, 'sc1', 'ok', None, None),
    (1, 'sc2', 'no', None, None),
    (1, 'sc3', 'yes', None, None),
    (1, 'sc4', None, '20180318', None),
    (1, 'sc5', None, None, '["5","2","9"]'),
    (1, 'sc6', None, '20180317', None),
    (1, 'sc7', 'ok++', None, None),
    (2, 'sc3', 'yes++', None, None),
    (2, 'sc2', 'no--', None, None),
    (2, 'sc7', 'ok--', None, None),
    (2, 'sc4', None, '20180315', None),
    (3, 'sc1', 'no', None, None),
    (3, 'sc6', None, '20180313', None)
]
df = spark.createDataFrame(data, schema=['id', 'k', 'v1', 'v2', 'v3'])

# 生成汇总行
summary_df = df.groupBy('id').agg(
    F.concat_ws(', ', F.collect_list(F.coalesce(F.col('v1'), F.lit('')))).alias('v1'),
    F.min(F.coalesce(F.col('v2'), F.lit('99999999'))).alias('v2'),
    F.concat_ws('; ', F.collect_list(F.coalesce(F.col('v3'), F.lit('')))).alias('v3')
).withColumn('k', F.lit('summary')).withColumn('row_type', F.lit('summary'))

# 合并原DataFrame和汇总行（自动对齐Schema）
final_df = df.withColumn('row_type', F.lit(None)).unionByName(summary_df)

# 查看结果
final_df.orderBy('id', 'k').show(truncate=False)

场景2：添加完全不同Schema的行 #

如果你的目标是添加和原DataFrame结构差异很大的行（比如新增统计类型的行，仅包含id、统计类型、统计值三列），需要先补全两边的缺失列（设为null），再合并：

Pandas 代码实现 #

# 构造完全不同结构的统计行
custom_stats = pd.DataFrame([
    (1, 'v1_non_null_count', 4),
    (2, 'v1_non_null_count', 3),
    (3, 'v1_non_null_count', 1)
], columns=['id', 'stat_type', 'stat_value'])

# 给原DataFrame补全新增列，给统计行补全原列
df_with_stats_cols = df.assign(stat_type=None, stat_value=None)
stats_with_original_cols = custom_stats.assign(k=None, v1=None, v2=None, v3=None)

# 合并得到最终结果
final_df = pd.concat([df_with_stats_cols, stats_with_original_cols], ignore_index=True)

PySpark 代码实现 #

# 构造完全不同结构的统计行
custom_stats_data = [
    (1, 'v1_non_null_count', 4),
    (2, 'v1_non_null_count', 3),
    (3, 'v1_non_null_count', 1)
]
custom_stats_df = spark.createDataFrame(custom_stats_data, schema=['id', 'stat_type', 'stat_value'])

# 补全两边的缺失列
df_with_stats_cols = df.withColumn('stat_type', F.lit(None)).withColumn('stat_value', F.lit(None))
stats_with_original_cols = custom_stats_df.withColumn('k', F.lit(None)) \
    .withColumn('v1', F.lit(None)) \
    .withColumn('v2', F.lit(None)) \
    .withColumn('v3', F.lit(None))

# 合并
final_df = df_with_stats_cols.unionByName(stats_with_original_cols)

关键注意事项 #

• Schema统一：合并前必须确保两个DataFrame的Schema完全一致，缺失列要补为null。
• 数据类型对齐：同一列的数据类型必须一致（比如原df的v2是字符串，新增行的v2也要是字符串，或者统一转换为日期类型）。
• 行标记：建议新增一个row_type列来区分原有行和新增行，方便后续筛选或处理。

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