Spark逐个加载指定文件并合并为单个DataFrame问题求助

嘿，作为Spark新手能写出单文件处理的代码已经很棒啦！针对你要逐个加载指定文件并合并成一个DataFrame的需求，我给你整理了两种实用的实现方式：

方法一：初始化空DataFrame，循环追加数据 #

这种方式贴合你习惯逐个处理文件的思路，核心是利用union方法将新读取的DataFrame合并到已有数据集中。划重点：所有文件的Schema必须完全一致，否则合并会报错。

示例代码如下（以Scala为例，Python逻辑完全一致，仅语法稍有差异）：

import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.types._

val spark = SparkSession.builder().appName("MergeFiles").getOrCreate()

// 1. 定义统一的Schema（如果文件自带表头，可省略这步用inferSchema，但显式定义更稳定）
val customSchema = StructType(Array(
  StructField("user_id", StringType, nullable = true),
  StructField("order_amount", DoubleType, nullable = true),
  // 按你的实际字段按需添加
))

// 2. 指定要加载的目标文件列表
val fileList = List(
  "/data/order_202401.csv",
  "/data/order_202402.parquet",
  "/data/order_202403.json"
)

// 3. 初始化空DataFrame（Schema必须和文件一致）
var mergedDF = spark.createDataFrame(spark.sparkContext.emptyRDD[Row], customSchema)

// 4. 循环逐个加载文件并合并
for (file <- fileList) {
  val tempDF = spark.read
    .schema(customSchema) // 应用统一Schema
    .format("csv") // 根据文件类型调整：parquet、json等
    .option("header", "true") // 文件有表头就保留，没有则删除
    .load(file)
  
  mergedDF = mergedDF.union(tempDF)
}

// 5. 合并完成后写入目标表
mergedDF.write
  .mode("overwrite") // 可选：overwrite覆盖已有表，append追加到已有表
  .saveAsTable("my_order_table")

方法二：收集所有DataFrame到列表，一次性合并 #

这种方式性能更优，减少了多次创建新DataFrame的开销，适合文件数量较多的场景：

示例代码：

import org.apache.spark.sql.SparkSession

val spark = SparkSession.builder().appName("MergeFiles").getOrCreate()

val fileList = List(
  "/data/order_202401.csv",
  "/data/order_202402.parquet"
)

// 1. 循环读取所有文件，生成DataFrame列表
val dfList = fileList.map(file => {
  spark.read
    .schema(customSchema) // 同样需要统一Schema
    .format("csv")
    .option("header", "true")
    .load(file)
})

// 2. 从列表中取出第一个DataFrame，依次合并剩余的DataFrame
val mergedDF = dfList.reduce((df1, df2) => df1.union(df2))

// 3. 写入目标表
mergedDF.write.mode("overwrite").saveAsTable("my_order_table")

关键注意事项 #

• Schema一致性：这是合并的核心前提，如果不同文件Schema不一致，建议先统一处理（比如补全缺失字段、转换字段类型）再合并。
• 文件格式匹配：确保format参数和你的文件类型一致（csv、parquet、json等），并添加对应的读取选项（比如csv的分隔符、json的多行模式）。
• 写入模式选择：mode("overwrite")会覆盖已有的目标表，如果需要保留原有数据并追加新数据，改用mode("append")。

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