Reading files dynamically from HDFS from within spark transformation functions

apache-spark, spark-streaming

Solution

You don't necessarily need service context to interact with HDFS. You can simply broadcast the hadoop configuration from master and use the broadcasted configuration value on executors to construct a `hadoop.fs.FileSystem`. Then the world is your. :)

Following is the code:

import java.io.StringWriter

import com.sachin.util.SparkIndexJobHelper._
import org.apache.commons.io.IOUtils
import org.apache.hadoop.conf.Configuration
import org.apache.hadoop.fs.{FileSystem, Path}
import org.apache.spark.rdd.RDD
import org.apache.spark.{SerializableWritable, SparkConf}

class Test {

  def main(args: Array[String]): Unit = {
    val conf = new SparkConf()
      .setMaster("local[15]")
      .setAppName("TestJob")
    val sc = createSparkContext(conf)

    val confBroadcast = sc.broadcast(new SerializableWritable(sc.hadoopConfiguration))

    val rdd: RDD[String] = ??? // your existing rdd
    val filedata_rdd = rdd.map { x => readFromHDFS(confBroadcast.value.value, x) }

  }

  def readFromHDFS(configuration: Configuration, path: String): String = {
    val fs: FileSystem = FileSystem.get(configuration)
    val inputStream = fs.open(new Path(path));

    val writer = new StringWriter();
    IOUtils.copy(inputStream, writer, "UTF-8");
    writer.toString();
  }

}

Problem

How can a file from HDFS be read in a spark function not using sparkContext within the function. Example: ``` val filedata_rdd = rdd.map { x => ReadFromHDFS(x.getFilePath) } ``` Question is how ReadFromHDFS can be implemented?Usually to read from HDFS we could do a sc.textFile but in this case sc cannot be used in the function.

Original source