title: Read multiple files from a directory using Spark
body: <p>I am trying to solve this <a href="https://www.kaggle.com/c/axa-driver-telematics-analysis" rel="nofollow">problem</a> at kaggle using spark:</p>
  
  <p>the hierarchy of input is like this :</p>
  
  <pre><code>drivers/{driver_id}/trip#.csv
  e.g., drivers/1/1.csv
        drivers/1/2.csv
        drivers/2/1.csv
  </code></pre>
  
  <p>I want to read the parent directory <strong>"drivers"</strong> and for each sub directory i would like to create a <strong>pairRDD</strong> with key as <strong>(sub_directory,file_name)</strong> and value as the content of the file</p>
  
  <p>I checked <a href="http://www.infoobjects.com/spark-core-sc-textfile-vs-sc-wholetextfiles/" rel="nofollow">this</a> link and tried to use   </p>
  
  <pre><code>val text = sc.wholeTextFiles("drivers")
  text.collect()
  </code></pre>
  
  <p>this failed with error :</p>
  
  <pre><code>java.lang.ArrayIndexOutOfBoundsException: 0
      at org.apache.hadoop.mapreduce.lib.input.CombineFileInputFormat$OneFileInfo.&lt;init&gt;(CombineFileInputFormat.java:591)
      at org.apache.hadoop.mapreduce.lib.input.CombineFileInputFormat.getMoreSplits(CombineFileInputFormat.java:283)
      at org.apache.hadoop.mapreduce.lib.input.CombineFileInputFormat.getSplits(CombineFileInputFormat.java:243)
      at org.apache.spark.rdd.WholeTextFileRDD.getPartitions(NewHadoopRDD.scala:267)
      at org.apache.spark.rdd.RDD$$anonfun$partitions$2.apply(RDD.scala:219)
      at org.apache.spark.rdd.RDD$$anonfun$partitions$2.apply(RDD.scala:217)
      at scala.Option.getOrElse(Option.scala:120)
      at org.apache.spark.rdd.RDD.partitions(RDD.scala:217)
      at org.apache.spark.SparkContext.runJob(SparkContext.scala:1779)
      at org.apache.spark.rdd.RDD$$anonfun$collect$1.apply(RDD.scala:885)
      at org.apache.spark.rdd.RDDOperationScope$.withScope(RDDOperationScope.scala:148)
      at org.apache.spark.rdd.RDDOperationScope$.withScope(RDDOperationScope.scala:109)
      at org.apache.spark.rdd.RDD.withScope(RDD.scala:286)
      at org.apache.spark.rdd.RDD.collect(RDD.scala:884)
  </code></pre>
  
  <p>but when i run the below code, it works.</p>
  
  <pre><code>val text =  sc.wholeTextFiles("drivers/1")
  text.collect()
  </code></pre>
  
  <p>but I don't want to do this, since here i will have to read the directory <strong>drivers</strong> and loop the files and call <strong>wholeTextFiles</strong> for each entry. </p>