在 Scala Spark 中加载 HDFS 上的任意二进制文件答案

【问题标题】：Load Arbitrary Binary Files on HDFS in Scala Spark在 Scala Spark 中加载 HDFS 上的任意二进制文件
【发布时间】：2018-02-17 05:16:41
【问题描述】：

场景：

HDFS 上有数千个海量二进制文件
def decode(String localFilePath): Array[MyCustomType] 可以根据文件的本地路径对文件进行解码

如何使用 Scala spark 并行加载这些文件并获得 RDD[MyCustomType] 作为回报？

PS。 decode 是一个 thrift 解码器，它获取 local 文件名，将 thrift 文件作为记录数组加载到内存中。

我认为这里缺少的难题是将文件从 HDFS 下载到节点并将本地名称传递给 decode..

【问题讨论】：

可能与此 stackoverflow.com/questions/37635065/… 关系模糊而遥远
感谢您的建议。更新了问题

标签： scala apache-spark hdfs

【解决方案1】：

解决方案是您需要使用 spark 的PortableDataStream 将 HDFS 上的二进制数据流式传输到计算节点。

val documentsRDD: RDD[Document] = sparkContext.binaryFiles("/data/path/on/hdfs")
  .flatMap { case (f: String, p: PortableDataStream) => {
    val stream: BufferedInputStream = new BufferedInputStream(new GZIPInputStream(dis), 2048)
    // you can take it from here and do the rest. in my case I was dealing with thrift:
    val protocol: TBinaryProtocol = new TBinaryProtocol(new TIOStreamTransport(stream))
  } 
}

【讨论】：