hadoop - 如何使用 Spark 从目录中递归读取 Hadoop 文件？

Question

在给定的目录中，我有许多不同的文件夹，在每个文件夹中，我都有 Hadoop 文件(part_001 等)。

directory
   -> folder1
      -> part_001...
      -> part_002...
   -> folder2
      -> part_001...
   ...

给定目录，我如何递归读取该目录中所有文件夹的内容，并使用 Scala 将这些内容加载到 Spark 中的单个 RDD 中？

我找到了这个，但它不会递归地进入子文件夹(我正在使用 import org.apache.hadoop.mapreduce.lib.input):

  var job: Job = null
  try {
    job = Job.getInstance()
    FileInputFormat.setInputPaths(job, new Path("s3n://" + bucketNameData + "/" + directoryS3))
    FileInputFormat.setInputDirRecursive(job, true)
  } catch {
    case ioe: IOException => ioe.printStackTrace(); System.exit(1);
  }
  val sourceData = sc.newAPIHadoopRDD(job.getConfiguration(), classOf[TextInputFormat], classOf[LongWritable], classOf[Text]).values

我也找到了这个web-page使用 SequenceFile，但我还是不明白如何将它应用到我的案例中？

Answer 1

如果您使用的是 Spark，则可以使用通配符来执行此操作，如下所示:

scala>sc.textFile("path/*/*")

sc 是 SparkContext，如果您使用的是默认初始化的 spark-shell，或者如果您正在创建自己的程序，则应该必须自己实例化一个 SparkContext。

小心以下标志:

scala> sc.hadoopConfiguration.get("mapreduce.input.fileinputformat.input.dir.recursive") 
> res6: String = null

你应该将这个标志设置为真:

sc.hadoopConfiguration.set("mapreduce.input.fileinputformat.input.dir.recursive","true")