Scala：如何从 RDD 中获取 PortableDataStream 实例的内容答案

【问题标题】：Scala: How to get the content of PortableDataStream instance from an RDDScala：如何从 RDD 中获取 PortableDataStream 实例的内容
【发布时间】：2020-02-04 06:56:31
【问题描述】：

因为我想从 binaryFiles 中提取数据，所以我使用 val dataRDD = sc.binaryRecord("Path") 我得到的结果是org.apache.spark.rdd.RDD[(String, org.apache.spark.input.PortableDataStream)]

我想提取我的文件内容，格式为PortableDataStream

为此我尝试过：val data = dataRDD.map(x => x._2.open()).collect() 但我收到以下错误： java.io.NotSerializableException:org.apache.hadoop.hdfs.client.HdfsDataInputStream

如果您知道如何解决我的问题，请帮忙！

非常感谢。

【问题讨论】：

标签： scala apache-spark rdd

【解决方案1】：

实际上，PortableDataStream 是可序列化的。这就是它的目的。然而，open() 返回一个简单的DataInputStream（在您的情况下为HdfsDataInputStream，因为您的文件位于 HDFS 上），它不可序列化，因此您会得到错误。

其实，当你打开PortableDataStream，你只需要马上读取数据。在scala中，你可以使用scala.io.Source.fromInputStream:

val data : RDD[Array[String]] = sc
    .binaryFiles("path/.../")
    .map{ case (fileName, pds) => {
        scala.io.Source.fromInputStream(pds.open())
            .getLines().toArray
    }}

此代码假定数据是文本的。如果不是，您可以调整它以读取任何类型的二进制数据。这是一个创建字节序列的示例，您可以按照自己的方式处理。

val rdd : RDD[Seq[Byte]] = sc.binaryFiles("...")
    .map{ case (file, pds) => {
        val dis = pds.open()
        val bytes = Array.ofDim[Byte](1024)
        val all = scala.collection.mutable.ArrayBuffer[Byte]()
        while( dis.read(bytes) != -1) {
            all ++= bytes
        }
        all.toSeq
    }}

请参阅javadoc of DataInputStream 了解更多可能性。比如拥有readLong、readDouble等方法。

【讨论】：

感谢您的回复。我已经尝试过您的解决方案，缺少“}”，我应该将.toSeq() 编辑为.toSeq，否则它会给我error: not enough arguments for method apply: (idx: Int)String in trait SeqLike。虽然解决方案给出了ERROR scheduler.TaskSetManager: Task 0 in stage 1.0 failed 1 times; aborting job
我没有测试所有的代码，抱歉。我修正了我的答案。
它实际上不适用于toSeq，可能是因为它只是简单地封装了基于inputStream的迭代器。 toArray 相反，实际上提取数据并将其放入数组中。我在答案中对其进行了修改。感谢您的反馈；）
感谢@Oli 的回复。问题依然存在java.nio.charset.MalformedInputException: Input length = 1
好吧，那不一样了。你在读什么样的数据？这只是字符串的一个例子。如果你读取二进制数据，它不会工作，你必须调整代码......

【解决方案2】：

val bf    = sc.binaryFiles("...")
val bytes = bf.map{ case(file, pds) => {
    val dis = pds.open()
    val len = dis.available();
    val buf = Array.ofDim[Byte](len)
    pds.open().readFully(buf)
    buf
}}
bytes: org.apache.spark.rdd.RDD[Array[Byte]] = MapPartitionsRDD[21] at map at <console>:26

scala> bytes.take(1)(0).size
res15: Int = 5879609  // this happened to be the size of my first binary file

【讨论】：