使用 Spark 流在 Hbase/HDFS 中保存 protobuf

Question

我希望使用 spark 流将 protobuf 消息存储在 Hbase/HDFS 中。我有以下两个问题

1. 存储大量protobuf的有效方法是什么 消息以及检索它们以执行某些操作的有效方法 分析？例如，它们应该存储为 Hbase 中的 Strings/byte[] 还是应该存储在 HDFS 等的 parquet 文件中。
2. protobuf的层次结构应该如何 消息被存储？我的意思是，嵌套元素是否应该展平 在存储之前取出，还是有任何机制可以按原样存储它们？ 如果嵌套元素是集合或映射，它们应该是 分解并存储为多行？

Protobuf 消息的示例结构如下所示

>     +--MsgNode-1
>       +--Attribute1 - String
>       +--Attribute2 - Int
>       +--MsgNode-2
>         +--Attribute1 - String
>         +--Attribute2 - Double
>         +--MsgNode-3 - List of MsgNode-3's
>           +--Attribute1 - Int

我打算使用 Spark 流来收集 protobuf 消息作为字节并将它们存储在 Hbase/HDFS 中。

Answer 1

问题 1：

存储大量protobuf消息的有效方法是什么 以及检索它们以进行一些分析的有效方法？为了 例如，它们应该在 Hbase 中存储为 Strings/byte[] 还是应该 它们存储在 HDFS 等的 parquet 文件中。

我会推荐 - 将 Proto-buf 存储为 Parquet AVRO 文件（使用 AVRO 模式拆分为有意义的消息）。

这可以使用 dataframes api spark 1.5 及更高版本来实现（PartiotionBy 和 SaveMode.Append）

看到这个a-powerful-big-data-trio

如果您存储为字符串或字节数组，则无法直接进行数据分析（查询原始数据）。

如果您使用的是cloudera，可以使用impala（支持parquet-avro）来查询原始数据。

问题 2：

protobuf消息的层次结构应该如何 存储？我的意思是，嵌套元素之前是否应该展平 存储，还是有任何机制可以按原样存储它们？如果嵌套 元素是集合或地图，如果它们被分解并存储为 多行？

如果您以原始格式存储来自 Spark Streaming 的数据，如果业务想要查询并知道他们收到了什么样的数据，您将如何查询（这个要求很常见）。

首先，您必须了解您的数据（即不同消息与 protobuf 之间的关系，以便您可以决定单行或多行）然后开发 protobuf 解析器来解析 protobuf 的消息结构。根据您的数据，将其转换为 avro 通用记录以保存为 parquet 文件。

提示：

protobuf 解析器可以根据您的要求以不同的方式开发。 其中一种通用方式如下例所示。

public SortedMap<String, Object> convertProtoMessageToMap(GeneratedMessage src) {

    final SortedMap<String, Object> finalMap = new TreeMap<String, Object>();
    final Map<FieldDescriptor, Object> fields = src.getAllFields();

    for (final Map.Entry<FieldDescriptor, Object> fieldPair : fields.entrySet()) {

        final FieldDescriptor desc = fieldPair.getKey();

        if (desc.isRepeated()) {
            final List<?> fieldList = (List<?>) fieldPair.getValue();
            if (fieldList.size() != 0) {
                final List<String> arrayListOfElements = new ArrayList<String>();
                for (final Object o : fieldList) {
                    arrayListOfElements.add(o.toString());
                }
                finalMap.put(desc.getName(), arrayListOfElements);
            }
        } else {

            finalMap.put(desc.getName(), fieldPair.getValue().toString());

        }

    }
    return finalMap;
}