Sauvegarde de protobuf dans Hbase/HDFS à l'aide du streaming Spark

Question

Je cherche à stocker les messages protobuf dans Hbase/HDFS en utilisant spark streaming. Et j'ai les deux questions suivantes

1. Quel est le moyen efficace de stocker un grand nombre de protobuf. et le moyen efficace de les récupérer pour faire de l'analyse ? d'analyse ? Par exemple, doivent-ils être stockés en tant que chaînes de caractères/byte[] dans Hbase ou doivent-ils être stockés dans des fichiers parquet dans HDFS, etc.
2. Comment la structure hiérarchique d'un protobuf doit-elle être stockée ? Je veux dire, les éléments imbriqués doivent-ils être aplatis avant d'être stockés, ou existe-t-il un mécanisme permettant de les stocker tels quels ? Si les éléments imbriqués sont des collections ou des cartes, doivent-ils être explosés et stockés sous forme de rangées multiples ?

L'exemple de structure d'un message Protobuf est présenté ci-dessous.

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

J'envisage d'utiliser le streaming Spark pour collecter les messages protobuf sous forme d'octets et les stocker dans Hbase/HDFS.

Answer 1

Question 1 :

Quelle est la manière efficace de stocker un grand nombre de messages protobuf ? et la manière efficace de les récupérer pour effectuer des analyses ? Pour par exemple, doivent-ils être stockés sous forme de chaînes de caractères/byte[] dans Hbase ou doivent-ils être stockés dans des fichiers parquet dans HDFS, etc.

Je recommande - de stocker le Proto-buf sous forme de fichiers Parquet AVRO (en le divisant en messages significatifs avec le schéma AVRO).

Ceci peut être réalisé en utilisant l'api dataframes de spark 1.5 et plus ( PartiotionBy con SaveMode.Append )

ver esto un-trio-de-grosse-donnée-puissant

Si vous stockez les données sous forme de chaîne ou de tableau d'octets, vous ne pouvez pas faire d'analyse de données directement (interrogation sur les données brutes).

Si vous utilisez cloudera, impala (qui prend en charge parquet-avro) peut être utilisé pour interroger rawdata.

Question 2 :

Comment la structure hiérarchique d'un message protobuf doit-elle être stockée ? Je veux dire, est-ce que les éléments imbriqués doivent être aplatis avant le stockage, ou existe-t-il un mécanisme permettant de les stocker tels quels ? Si les éléments imbriqués sont des collections ou des cartes, doivent-ils être éclatés et stockés en tant que rangées multiples ?

Si vous stockez les données dans un format brut à partir de spark streaming, comment allez-vous les interroger si l'entreprise veut les interroger et savoir quel type de données elle a reçu (cette exigence est très courante).

En premier lieu, vous devez comprendre vos données (c.-à-d. la relation entre les différents messages dans protobuf de sorte qu'une seule ligne ou plusieurs lignes, vous pouvez décider) puis développer un analyseur de protobuf pour analyser la structure des messages de protobuf. sur la base de vos données, convertissez-le en enregistrement générique avro pour le sauvegarder comme fichier parquet.

TIP :

Les analyseurs syntaxiques de protobuf peuvent être développés de différentes manières en fonction de vos besoins. L'une des méthodes génériques est l'exemple ci-dessous.

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;
}