hadoop - 在 RDD 方法中访问 HDFS 输入拆分路径

Question

我的 HDFS 文件路径包含我想在 Spark 中访问的元数据，即类似以下内容的内容:

sc.newAPIHadoopFile("hdfs://.../*"), ...)
  .map( rdd => /* access hdfs path here */ )

在 Hadoop 中，我可以通过 FileSplit.getPath() 访问整个拆分的路径。我可以在 Spark 中做任何类似的事情吗，或者我是否必须将路径字符串附加到扩展 NewHadoopRDD 中的每个 RDD 元素，我认为这可能相当昂贵？

Answer 1

在您提供给 map() 方法的闭包中，没有可用的元数据/执行上下文信息。

你可能想要的是

mapPartitionsWithContext

Similar to mapPartitions, but allows accessing information about the processing state within the mapper

然后你可以做类似的事情

import org.apache.spark.TaskContext
def myfunc(tc: TaskContext, iter: Iterator[Int]) : Iterator[Int] = {
  tc.addOnCompleteCallback(() => println(
    "Partition: "     + tc.partitionId +
    ", AttemptID: "   + tc.attemptId   +
    ", Interrupted: " + tc.interrupted))

  iter.toList.filter(_ % 2 == 0).iterator
}
a.mapPartitionsWithContext(myfunc).collect

更新 以前的解决方案不提供 HDFS 文件名。你可能需要做这样的事情:

• Create a custom InputFormat that extends the FileInputFormat
• Create a custom RecordReader that for each line outputs the file associated with the InputSplit and then the actual value for each line
• In your spark mapper you will parse out the first field that now caontains the hdfs filename, and the rest of the mapper remains the same