如何用Scala+Apache Spark将目录下多份文本数据集合并为单个DataFrame

嘿，这个需求在大数据日常开发里太常见了！我来给你分享几种靠谱的实现方式，都是实战中验证过的：

方法1：直接读取整个目录（最便捷） #

Spark天生支持读取目录下的所有同结构文件，只要你的文本文件列名、类型完全一致，一行代码就能搞定。

代码示例 #

import org.apache.spark.sql.SparkSession

// 初始化SparkSession（本地测试用master("local[*]")，生产环境请移除该配置）
val spark = SparkSession.builder()
  .appName("MergeTextDatasets")
  .master("local[*]")
  .getOrCreate()

// 读取目录下所有文本文件，自动合并为DataFrame
val mergedDF = spark.read
  .option("header", "true") // 如果你的文件有表头就加这个，没有则删除
  .option("inferSchema", "true") // 自动推断列类型（生产环境更推荐手动指定Schema）
  .csv("/path/to/your/target-directory") // 注意：如果是分隔符结构化文本用csv；纯每行一条记录用text方法

// 验证结果
mergedDF.printSchema()
mergedDF.show(5)

小提示 #

• 如果你的文本是无分隔符的纯文本行（比如每行是一条完整的字符串记录），把.csv()换成.text()即可，得到的DataFrame会有一个名为value的列。
• 目录下的子目录也会被递归读取，如果不想递归，加.option("recursiveFileLookup", "false")。

方法2：手动指定Schema（生产环境推荐） #

inferSchema虽然方便，但大数据量下会触发额外的扫描，而且可能推断出错误的类型（比如把数字字符串误判为String）。生产环境强烈建议手动定义Schema，保证稳定性和性能。

代码示例 #

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

// 定义与你的数据集完全匹配的Schema
val customSchema = StructType(Seq(
  StructField("user_id", IntegerType, nullable = false),
  StructField("user_name", StringType, nullable = true),
  StructField("register_date", DateType, nullable = true),
  StructField("score", DoubleType, nullable = true)
))

val mergedDF = spark.read
  .option("header", "true")
  .schema(customSchema) // 绑定手动定义的Schema
  .option("delimiter", "\t") // 如果是制表符分隔，替换成你的分隔符（比如","、"|"）
  .csv("/path/to/your/target-directory")

方法3：灵活匹配特定文件 #

如果不需要合并目录下所有文件，而是要匹配特定文件名（比如前缀、后缀匹配），可以用通配符：

// 合并目录下所有以"user_data_"开头的txt文件
val mergedDF = spark.read
  .option("header", "true")
  .schema(customSchema)
  .csv("/path/to/your/directory/user_data_*.txt")

// 或者指定多个具体文件路径
val mergedDF = spark.read
  .option("header", "true")
  .schema(customSchema)
  .csv(
    "/path/to/file1.txt",
    "/path/to/file2.txt",
    "/path/to/subdir/file3.txt"
  )

常见细节处理 #

• 编码问题：如果文件不是UTF-8编码，添加.option("encoding", "GBK")（替换成你的编码格式）。
• 空值处理：可以加.option("nullValue", "NA")指定空值标记，让Spark正确识别空值。
• 跳过损坏文件：添加.option("mode", "DROPMALFORMED")自动跳过格式错误的行，或者"FAILFAST"直接报错终止（根据业务需求选择）。

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