如何使用 Apache Spark 读取/写入协议缓冲区消息？答案

【问题标题】：How to read/write protocol buffer messages with Apache Spark?如何使用 Apache Spark 读取/写入协议缓冲区消息？
【发布时间】：2019-02-05 08:53:33
【问题描述】：

我想使用 Apache Spark 从/向 HDFS 读取/写入协议缓冲区消息。我找到了这些建议的方法：

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

2) 转换为 Parquet 文件。这种方式有parquet-mr 和sparksql-protobuf github 项目，但我不想要parquet 文件，因为我总是使用所有列（不是某些列），这样Parquet 格式不会给我任何收益（至少我想想）。

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

4）通过使用序列文件，这是我正在寻找的，但没有发现任何相关内容。所以，我的问题是：如何将 protobuf 消息写入 HDFS 上的序列文件并从中写入？任何其他建议都会很有用。

5) 通过 Twitter 的 Elephant-bird 库。

【问题讨论】：

标签： apache-spark hdfs protocol-buffers sequencefile

【解决方案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

【讨论】：