title: How can I read HDFS files from a spark executor?
body: <p>I have a large (> 500m row) CSV file. Each row in this CSV file contains a path to a binary file located on HDFS. I want to use Spark to read each of these files, process them, and write out the results to another CSV file or a table.</p>
  
  <p>Doing this is simple enough in the driver, and the following code gets the job done</p>
  
  <pre><code>val hdfsFilePathList = // read paths from CSV, collect into list
  
  hdfsFilePathList.map( pathToHdfsFile =&gt; {
    sqlContext.sparkContext.binaryFiles(pathToHdfsFile).mapPartitions { 
      functionToProcessBinaryFiles(_)
    }
  })
  </code></pre>
  
  <p>The major problem with this is that the driver is doing too much of the work. I would like to farm out the work done by <code>binaryFiles</code> to the executors. I've found some promising examples that I thought would allow me to access the sparkContext from an executor:</p>
  
  <p><a href="https://stackoverflow.com/questions/38224132/use-sparkcontext-hadoop-configuration-within-rdd-methods-closures-like-foreachp">Use SparkContext hadoop configuration within RDD methods/closures, like foreachPartition</a></p>
  
  <p><a href="https://github.com/apache/spark/blob/master/core/src/main/scala/org/apache/spark/util/SerializableConfiguration.scala" rel="nofollow noreferrer">https://github.com/apache/spark/blob/master/core/src/main/scala/org/apache/spark/util/SerializableConfiguration.scala</a></p>
  
  <p>but they don't seem to work the way I thought they would. I'd expect the following to work:</p>
  
  <pre><code>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 =&gt; {
      serConf.get()
  })
  </code></pre>
  
  <p>But it fails with <code>KryoException: java.util.ConcurrentModificationException</code></p>
  
  <p>Is it possible to have to executors access HDFS files or the HDFS file system directly? Or alternatively, is there an efficient way to read millions of binary files on HDFS/S3 and process them with Spark?</p>