apache-spark - 如何使用 Apache Spark 读/写 Protocol Buffer 消息？

Question

我想使用 Apache Spark 从/向 HDFS 读取/写入 Protocol Buffer 消息。我找到了这些建议的方法:

1) 使用 Google 的 Gson 库将 protobuf 消息转换为 Json，然后通过 SparkSql 读取/写入它们。此解决方案在 this link 中有解释但我认为这样做(转换为 json)是一项额外的任务。

2) 转换为 Parquet 文件。有parquet-mr和 sparksql-protobuf github 项目以这种方式，但我不想要 Parquet 文件，因为我总是使用所有列(而不是某些列)，这样 Parquet Format 不会给我任何好处(至少我认为)。

3) ScalaPB .可能这就是我要找的。但是在Scala语言中，我对此一无所知。我正在寻找基于 Java 的解决方案。 This youtube video介绍 scalaPB 并解释如何使用它(针对 Scala 开发人员)。

4)通过使用序列文件，这就是我要找的，但一无所获。所以，我的问题是:如何将 protobuf 消息写入 HDFS 上的序列文件？任何其他建议都会有用。

5) 通过推特的 Elephant-bird图书馆。

Answer 1

虽然有点隐藏在点之间，但您似乎在问如何在 spark 中写入序列文件。我找到了一个例子 here .

// Importing org.apache.hadoop.io package
import org.apache.hadoop.io._

// As we need data in sequence file format to read. Let us see how to write first
// Reading data from text file format
val dataRDD = sc.textFile("/public/retail_db/orders")

// Using null as key and value will be of type Text while saving in sequence file format
// By Int and String, we do not need to convert types into IntWritable and Text
// But for others we need to convert to writable object
// For example, if the key/value is of type Long, we might have to 
// type cast by saying new LongWritable(object)
dataRDD.
  map(x => (NullWritable.get(), x)).
  saveAsSequenceFile("/user/`whoami`/orders_seq")
// Make sure to replace `whoami` with the appropriate OS user id

// Saving in sequence file with key of type Int and value of type String
dataRDD.
  map(x => (x.split(",")(0).toInt, x.split(",")(1))).
  saveAsSequenceFile("/user/`whoami`/orders_seq")
// Make sure to replace `whoami` with the appropriate OS user id