能否在Spark Executor中读取HDFS文件？附需求与代码示例

你遇到的问题核心是SparkContext（SC）是Driver端专属对象，无法在Executor的mapPartitions算子内正常使用：

• SC无法被序列化传递到Executor，强行传递会报序列化错误；
• 集群环境下Executor不允许初始化新的SparkContext，这会破坏Spark的集群架构。

下面提供两种可行的实现方式：

方式一：直接使用Hadoop FileSystem API读取HDFS #

通过Hadoop原生的FileSystem API，在Executor端直接读取HDFS文件，不需要依赖SparkContext。

代码示例 #

import org.apache.hadoop.fs.{FileSystem, Path}
import org.apache.spark.{SparkConf, SparkContext}

object SparkTest2 {
  def main(args: Array[String]): Unit = {
    val conf = new SparkConf().setAppName("SparkTest")
    val sc = new SparkContext(conf)
    // 获取Hadoop配置，后续传递给Executor
    val hadoopConf = sc.hadoopConfiguration
    // 广播Hadoop配置，避免每个任务重复传递
    val broadcastConf = sc.broadcast(hadoopConf)

    val rdd = sc.textFile("test1")
    val resultRDD = rdd.mapPartitions { partitionIter =>
      // 从广播变量获取Hadoop配置
      val conf = broadcastConf.value
      // 获取HDFS FileSystem实例
      val fs = FileSystem.get(conf)
      // 要读取的HDFS文件路径
      val filePath = new Path("/path/to/your/hdfs/file")
      
      // 读取文件内容
      val inputStream = fs.open(filePath)
      val lines = scala.io.Source.fromInputStream(inputStream).getLines()
      
      // 结合partitionIter做业务处理，比如关联分区数据和读取的文件内容
      val processed = partitionIter.flatMap { partitionLine =>
        lines.map(fileLine => s"partition data: $partitionLine | file data: $fileLine")
      }
      
      // 关闭资源
      inputStream.close()
      fs.close()
      
      processed
    }

    resultRDD.collect().foreach(println)
    sc.stop()
  }
}

注意事项 #

• 确保项目依赖中包含Hadoop相关包（比如hadoop-client）；
• 每次使用完FileSystem和输入流后要手动关闭，避免资源泄漏；
• 用广播变量传递Hadoop配置，减少序列化开销。

方式二：用Spark的分区关联算子（zipPartitions） #

如果你的需求是每个分区对应读取一个特定的HDFS文件，可以先将目标文件集创建为RDD，再通过zipPartitions将原RDD和文件RDD按分区关联，这种方式更符合Spark的分布式计算范式。

代码示例 #

import org.apache.spark.{SparkConf, SparkContext}

object SparkTest2 {
  def main(args: Array[String]): Unit = {
    val conf = new SparkConf().setAppName("SparkTest")
    val sc = new SparkContext(conf)

    // 原数据RDD
    val dataRDD = sc.textFile("test1")
    // 要读取的文件集RDD（确保分区数和dataRDD一致）
    val fileRDD = sc.textFile("/path/to/hdfs/files/*").repartition(dataRDD.getNumPartitions)

    // 按分区关联两个RDD，处理数据
    val resultRDD = dataRDD.zipPartitions(fileRDD) { (dataIter, fileIter) =>
      // 同时遍历分区数据和对应文件的内容
      dataIter.flatMap(dataLine => fileIter.map(fileLine => s"data: $dataLine | file: $fileLine"))
    }

    resultRDD.collect().foreach(println)
    sc.stop()
  }
}

注意事项 #

• 确保两个RDD的分区数一致，否则zipPartitions会报错；
• 如果文件数量和原RDD分区数不匹配，可以先调整分区数（比如repartition）。

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