如何从spark Executor中读取HDFS文件？

Question

我有一个大的(>500M行）CSV文件。这个CSV文件中的每一行都包含一个位于HDFS上的二进制文件的路径。我想使用Spark读取这些文件中的每一个，处理它们，并将结果写到另一个CSV文件或表中。

在驱动程序中执行此操作非常简单，下面的代码完成了这项工作

val hdfsFilePathList = // read paths from CSV, collect into list

hdfsFilePathList.map( pathToHdfsFile => {
  sqlContext.sparkContext.binaryFiles(pathToHdfsFile).mapPartitions { 
    functionToProcessBinaryFiles(_)
  }
})

import java.io.{ObjectInputStream, ObjectOutputStream}
import org.apache.hadoop.fs.FileSystem
import org.apache.hadoop.conf.Configuration

class ConfigSerDeser(var conf: Configuration) extends Serializable {

  def this() {
    this(new Configuration())
  }

  def get(): Configuration = conf

  private def writeObject (out: java.io.ObjectOutputStream): Unit = {
    conf.write(out)
  }

  private def readObject (in: java.io.ObjectInputStream): Unit = {
    conf = new Configuration()
    conf.readFields(in)
  }

  private def readObjectNoData(): Unit = {
    conf = new Configuration()
  }
}

val serConf = new ConfigSerDeser(sc.hadoopConfiguration)

val mappedIn = inputDf.map( row => {
    serConf.get()
})

但是kryoException:java.util.concurrentModificationException失败

是否可以让执行者直接访问HDFS文件或HDFS文件系统？或者，是否有一种有效的方法来读取HDFS/S3上数百万个二进制文件并用Spark处理它们？

Answer 1

有一个类似的用例，我试图做同样的事情，但意识到SparkSession或SparkContext是不可序列化的，因此不能从执行器访问。