Spark Scala实现：读取不同Schema的Parquet文件并分路径输出

核心思路 #

由于每个子文件夹下的Parquet文件Schema不同，无法统一读取后批量处理，因此必须逐个遍历子文件夹，独立完成读取、转换、写入操作，以此保留原有的目录层级结构。

实现步骤 #

1. 初始化SparkSession：配置基础Spark参数，确保支持Parquet格式的读写能力。
2. 获取子文件夹列表：通过Hadoop的FileSystem API遍历输入父文件夹，筛选出所有子目录（避免直接读取文件）。
3. 逐个处理子文件夹：
   • 提取子文件夹名称，构造对应的输入、输出路径。
   • 读取当前子文件夹下的Parquet文件（Spark会自动解析文件的Schema）。
   • 执行自定义转换逻辑（根据业务需求处理DataFrame）。
   • 将转换后的DataFrame写入对应输出子文件夹，保持原文件结构。

完整代码示例 #

import org.apache.spark.sql.SparkSession
import org.apache.hadoop.fs.{FileSystem, Path}

object ParquetFolderProcessor {
  def main(args: Array[String]): Unit = {
    // 初始化SparkSession
    val spark = SparkSession.builder()
      .appName("ParquetFolderProcessor")
      .master("local[*]") // 生产环境可移除，由集群配置决定
      .getOrCreate()

    import spark.implicits._

    // 配置输入输出父路径
    val inputParentPath = "/path/to/Parent_input_folder"
    val outputParentPath = "/path/to/Parent_output_folder"

    // 获取Hadoop FileSystem实例，适配分布式存储
    val fs = FileSystem.get(spark.sparkContext.hadoopConfiguration)
    val parentPath = new Path(inputParentPath)

    // 筛选父文件夹下的所有子目录
    val childFolders = fs.listStatus(parentPath)
      .filter(_.isDirectory)
      .map(_.getPath)

    // 遍历处理每个子文件夹
    childFolders.foreach(childPath => {
      val childFolderName = childPath.getName
      val outputChildPath = s"$outputParentPath/$childFolderName"

      // 读取Parquet文件，自动识别Schema
      val rawDf = spark.read.parquet(childPath.toString)

      // --------------------------
      // 自定义转换逻辑：替换为你的业务处理代码
      val transformedDf = rawDf
        .withColumn("processed_time", current_timestamp()) // 示例转换：添加处理时间列
        // 更多转换操作...
      // --------------------------

      // 写入输出目录，保持原结构
      transformedDf.write
        .mode("overwrite") // 根据需求选择模式：append/overwrite/ignore
        .parquet(outputChildPath)

      println(s"处理完成：$childFolderName -> $outputChildPath")
    })

    // 关闭资源
    fs.close()
    spark.stop()
  }
}

关键注意事项 #

• 分布式存储适配：如果运行在HDFS/S3等分布式环境，输入输出路径需使用对应协议（如hdfs://、s3://），FileSystem会自动适配存储类型。
• Schema兼容性：Spark读取Parquet时会自动解析文件Schema，无需手动指定，天然适配不同子文件的结构差异。
• 写入模式选择：overwrite会覆盖已有输出目录，append会追加数据，ignore会跳过已存在的目录，可根据业务需求调整。
• 性能优化：若子文件夹数量极多，可调整Spark并行度参数（如spark.sql.shuffle.partitions），或考虑异步处理（需注意资源占用）。

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