使用Spark并行处理多文件且不合并的技术问题求助

问题根源 #

你当前代码用的是Scala本地集合的map方法，所有文件处理逻辑都在Driver端串行执行，完全没有分发到Executor节点，这就是任务无法并行的核心原因。另外，sc.textFile的编码问题可以通过Spark CSV读取器的encoding参数直接解决。

解决方案 #

要实现每个Executor处理一个文件、单个文件利用多核并行处理，需要做以下核心调整：

1. 将本地文件名列表转为Spark分布式RDD #

只有基于Spark的RDD/DataFrame，才能触发分布式任务调度，把文件处理逻辑真正分发到Executor节点执行。

2. 配置Spark读取参数解决编码与分区问题 #

通过CSV读取器的encoding参数指定文件编码，同时设置numPartitions强制单个大文件拆分为多个分区，充分利用Executor的多核资源。

3. 匹配Spark资源配置需求 #

调整Executor核数、实例数，确保每个Executor能承接单个文件的多分区处理任务。

修改后的完整代码 #

import org.apache.spark.sql.SparkSession

// 初始化SparkSession，配置资源适配需求
val spark = SparkSession.builder()
  .appName("TSVtoParquetConverter")
  .config("spark.executor.cores", "3")       // 每个Executor分配3核
  .config("spark.task.cpus", "1")            // 每个任务占用1核，让单个Executor可并行3个任务
  .config("spark.executor.instances", full_filelist.size) // Executor实例数等于文件总数
  .getOrCreate()

// 定义分布式文件处理函数（确保所有参数可序列化）
def processSingleFile(
  fileName: String,
  outFileDir: String,
  mountedSource1: String,
  mountedSource2: String,
  dominoSource: String
): Unit = {
  // 安全拼接输入输出路径
  val inFileDir = if (dominoSource.contains(fileName)) mountedSource1 else mountedSource2
  val inFullPath = s"$inFileDir/$fileName"
  val fileOutputName = fileName.replace(".txt.gz", "")
  val outFullPath = s"$outFileDir/$fileOutputName"

  // 读取TSV文件、指定编码与分区数，写入Parquet
  spark.read
    .option("sep", "\t")
    .option("header", "true")
    .option("encoding", "UTF-8") // 替换为文件实际编码（如GBK、ISO-8859-1等）
    .option("numPartitions", 3)  // 强制将单个文件拆分为3个分区
    .csv(inFullPath)
    .write
    .mode("overwrite") // 根据需求选择写入模式：append/ignore/error
    .parquet(outFullPath)
}

// 将本地文件名列表转为分布式RDD，并行度设为文件数量（每个文件对应一个RDD分区）
val fileRDD = spark.sparkContext.parallelize(full_filelist, full_filelist.size)

// 分发任务到Executor执行
fileRDD.foreach(fileName => 
  processSingleFile(
    fileName,
    mounted_target_filepath,
    mounted_source_filepath1,
    mounted_source_filepath2,
    domino_source
  )
)

关键调整说明 #

• 分布式执行触发：用spark.sparkContext.parallelize将本地列表转为RDD，每个文件对应一个RDD分区，Spark会自动将分区分配到不同Executor节点。
• 编码问题解决：通过encoding参数直接指定文件实际编码，彻底替代sc.textFile的编码局限性。
• 多核资源利用：设置numPartitions=3让单个大文件拆分为3个分区，配合spark.executor.cores=3与spark.task.cpus=1，单个Executor的3核可同时处理该文件的3个分区。
• 路径安全拼接：用字符串插值s"$path/$fileName"替代concat，避免路径拼接时的分隔符缺失问题。

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