在Spark中如何从多个不同格式的文件创建统一DataFrame？

可以实现，你提到的两种思路都有明确的适用边界，并不适配当前需求：

• JOIN操作是按关联字段做列级别的匹配拼接，如果你的需求是把CSV和JSON两类文件的数据按行合并到同一个DataFrame，JOIN完全不满足要求
• 先统一转成同格式再加载会产生额外的磁盘IO开销，数据量大的时候会浪费大量计算、存储资源，不推荐

核心实现逻辑 #

分别加载两种格式的文件为DataFrame，对齐Schema后直接按行合并即可，无需额外的格式转换操作。

情况1：两类文件Schema完全一致 #

直接使用unionByName合并即可，相比普通union按位置对齐列的逻辑，unionByName按列名对齐，避免列顺序不一致导致的数据错乱：

# 加载CSV文件，可根据实际情况调整header、分隔符等参数
csv_df = spark.read.csv("/path/to/csv_dir/*", header=True, inferSchema=True)
# 加载JSON文件
json_df = spark.read.json("/path/to/json_dir/*")

# 按列名合并，allowMissingColumns参数（Spark3.1+支持）可自动将某类文件缺失的列填充为null
unified_df = csv_df.unionByName(json_df, allowMissingColumns=True)

如果你使用的Spark版本低于3.1，手动为缺失列补全null值即可：

from pyspark.sql.functions import lit

# 示例：给json_df补全csv_df中存在的extra_col列
json_df = json_df.withColumn("extra_col", lit(None).cast("string"))
unified_df = csv_df.unionByName(json_df)

情况2：两类文件Schema不一致 #

先对两类DataFrame做列名、字段类型的标准化对齐，再执行合并操作即可：

# 示例：将json_df的列调整为和csv_df完全一致
json_aligned_df = json_df.selectExpr(
    "cast(uid as string) as user_id", # 调整字段类型+重命名
    "user_name as username", # 重命名列
    "cast(reg_ts as timestamp) as register_time" # 转换字段类型
)
# 合并
unified_df = csv_df.unionByName(json_aligned_df)

优化建议 #

• 生产环境建议提前明确定义统一的Schema，不要依赖inferSchema自动推断，既可以避免类型推断错误，又能大幅提升文件加载速度
• 如果两类文件存在重复数据，合并后可通过unified_df.dropDuplicates(["唯一标识列"])快速去重
• 如果需要频繁加载混合格式的数据源，可以把上述逻辑封装为通用工具函数，调用时直接传入两类文件的路径即可使用

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