如何使用Apache Avro序列化JSON文档然后写入Cassandra？

Question

这些天我读了很多关于

Apache Avro

的内容，我更倾向于使用它而不是使用

JSON

。目前，我们正在做的是，我们使用

JSON

序列化

Jackson

文档，然后将每个

JSON

的序列化

Cassandra

文档写入

row key/user id

中。然后我们有一个 REST 服务，它使用行键读取整个

JSON

文档，然后反序列化并进一步使用它。

我们将这样写入 Cassandra -

user-id   column-name   serialize-json-document-value

下面是一个示例，显示了我们正在写入 Cassandra 的 JSON 文档。该 JSON 文档适用于特定的行键/用户 ID。

{
  "lv" : [ {
    "v" : {
      "site-id" : 0,
      "categories" : {
        "321" : {
          "price_score" : "0.2",
          "confidence_score" : "0.5"
        },
        "123" : {
          "price_score" : "0.4",
          "confidence_score" : "0.2"
        }
      },
      "price-score" : 0.5,
      "confidence-score" : 0.2
    }
  } ],
  "lmd" : 1379214255197
}

现在我们正在考虑使用 Apache Avro，以便我们可以通过使用 Apache Avro 序列化来压缩此 JSON 文档，然后将其存储在 Cassandra 中。我对此有几个问题-

1. 是否可以首先使用 Apache Avro 序列化上述 JSON 文档，然后将其写入 Cassandra 中？如果是，我该怎么做？谁能提供一个简单的例子吗？
2. 我们还需要在从 REST 服务中从 Cassandra 读回它时对其进行反序列化。这也可以吗？

下面是我的简单代码，它序列化 JSON 文档并将其打印在控制台上。

public static void main(String[] args) {

    final long lmd = System.currentTimeMillis();

    Map<String, Object> props = new HashMap<String, Object>();
    props.put("site-id", 0);
    props.put("price-score", 0.5);
    props.put("confidence-score", 0.2);

    Map<String, Category> categories = new HashMap<String, Category>();
    categories.put("123", new Category("0.4", "0.2"));
    categories.put("321", new Category("0.2", "0.5"));
    props.put("categories", categories);

    AttributeValue av = new AttributeValue();
    av.setProperties(props);

    Attribute attr = new Attribute();
    attr.instantiateNewListValue();
    attr.getListValue().add(av);
    attr.setLastModifiedDate(lmd);

    // serialize it
    try {
        String jsonStr = JsonMapperFactory.get().writeValueAsString(attr);

        // then write into Cassandra
        System.out.println(jsonStr);
    } catch (JsonGenerationException e) {
        e.printStackTrace();
    } catch (JsonMappingException e) {
        e.printStackTrace();
    } catch (IOException e) {
        e.printStackTrace();
    }
}

Serialzie JSON 文档看起来像这样 -

{"lv":[{"v":{"site-id":0,"categories":{"321":{"price_score":"0.2","confidence_score":"0.5"},"123":{"price_score":"0.4","confidence_score":"0.2"}},"price-score":0.5,"confidence-score":0.2}}],"lmd":1379214255197}

AttributeValue

和

Attribute

类正在使用

Jackson Annotations

。

还有一个重要的注意事项，上述 json 文档中的属性将根据列名称进行更改。不同的列名有不同的属性。有些列名有两个属性，有些有 5 个属性。因此，根据我们拥有的元数据，上述 JSON 文档将具有其正确的属性和值。

我希望问题足够清楚。任何人都可以为此提供一个简单的示例，我如何使用 Apache Avro 来实现这一点。我刚刚开始使用 Apache Avro，所以我遇到了很多问题..

Answer 1

既然您已经使用了 jackson，您可以尝试使用 Jackson 数据格式模块来支持 Avro 编码数据。

Answer 2

Avro需要一个模式，所以你必须在使用它之前设计它；并且用法与自由格式的 JSON 有很大不同。

但是您可能需要考虑Smile，而不是 Avro，这是一种一对一的 JSON 二进制序列化，专为您可能想要在 JSON 和二进制数据之间来回切换的用例而设计；例如，使用 JSON 进行调试，或者在为 Javascript 客户端提供服务时。

Jackson 有 Smile 后端（请参阅https://github.com/FasterXML/jackson-dataformat-smile），实际上只需一行更改即可使用 Smile 代替（或补充）JSON。 很多项目都使用它（例如Elastic Search），并且它是成熟稳定的格式； Jackson 提供的工具支持广泛适用于不同的数据类型。

Answer 3

我尝试过使用 ObjectMapper 和 Gson，但它在所有情况下都不太适合我，所以我为此使用了 DatumWriter 和 DatumReader：

public static <T extends GenericRecord> String convertAvroObjectToJsonString(T event) throws IOException {
    try {
        DatumWriter<T> writer = new SpecificDatumWriter<>(event.getSchema());
        OutputStream out = new ByteArrayOutputStream();
        JsonEncoder encoder = EncoderFactory.get()
                .jsonEncoder(event.getSchema(), out);
        writer.write(event, encoder);
        encoder.flush();
        return out.toString();
    } catch (IOException e) {
        log.error("IOException occurred.", e);
        throw e;
    }
}

public static <T extends GenericRecord> T convertStringToAvro(String content, Schema schema) throws IOException {
    try {
        DatumReader<T> reader = new SpecificDatumReader<>(schema);
        JsonDecoder encoder = DecoderFactory.get()
                .jsonDecoder(schema, content);
        return reader.read(null, encoder);
    } catch (IOException e) {
        log.error("IOException occurred.", e);
        throw e;
    }
}

要获取反序列化的模式，我只需调用 MyObjectAvroClass.getClassSchema();