如何在Apache Kafka中使用AVRO序列化处理嵌套源数据？

Question

我的目标是从HTTP源获取JSON数据并使用AVRO序列化将其存储在Kafka主题中。

使用Kafka Connect和一个HTTP源连接器以及一堆SMT，我成功地创建了一个连接数据结构，当使用StringConverter写入主题时，它是这样的：

结构{base=stations，cod=200，coord=Struct{lat=54.0，lon=9.0}，dt=163210605}

因此，JSON被成功解析为结构，我可以使用SMT操作单个元素。接下来，我在Confluent schema注册表中创建了一个具有相应模式的新主题，并将连接器的值转换器切换到Confluent AVRO转换器，使用“value.converter”：“io.Confluent.connect.AVRO.AvroConverter”。

我收到的不是预期的序列化，而是一条错误消息：

org.apache.kafka.common.errors.SerializationException：错误序列化Avro消息引起的：org.apache.avro.SchemaParseException：无法重新定义：io.confluent.connect.avro.ConnectDefault

只要我用ReplaceField删除嵌套结构，或者用flatte简化结构，AVRO序列化就会像一个魔咒一样工作。因此，转换器似乎无法处理嵌套结构。

当您拥有嵌套元素并希望它们被序列化而不是将JSON存储为字符串并尝试在消费者或其他地方处理对象创建时，正确的方法是什么？这在Kafka Connect中可能吗？

Answer 1

从JSON字符串创建结构元素可以通过不同的方式实现。最初，使用SMT ExpandJson是为了简单。然而，它没有创建足够命名的结构，因为它没有一个模式可供使用。这就是AVRO序列化程序使用泛型类io时产生初始错误消息的原因。汇合的。连接阿夫罗。对于这些结构，ConnectDefault，如果存在多个结构，则存在歧义，这会引发异常。

另一个似乎做同样事情的SMT是Json模式，它有一个记录在案的FromJson转换。它确实接受模式，因此绕过了ExpandJson将嵌套元素解析为泛型类型的问题。不过，现在接受的是JSON模式，而到AVRO全名的映射是通过将“properties”作为名称空间并复制字段名来实现的。在本例中，您将得到属性。coord作为内部元素的全名。

例如，将以下JSON模式传递给SMT时：

{
  "$schema": "http://json-schema.org/draft-04/schema#",
  "type": "object",
  "properties": {
    "coord": {
      "type": "object",
      "properties": {
        "lon": {
          "type": "number"
        },
        "lat": {
          "type": "number"
        }
      },
      "required": [
        "lon",
        "lat"
      ]
    },
    ...
}

它生成的AVRO模式（因此在模式注册表中查找）是：

{
    "type": "record",
    "fields": [
        ...
        {
            "name": "coord",
            "type": {
                "type": "record",
                "name": "coord",
                "namespace": "properties",
                "fields": [
                    {
                        "name": "lat",
                        "type": "double"
                    },
                    {
                        "name": "lon",
                        "type": "double"
                    }
                ],
                "connect.name": "properties.coord"
            }
        },
    ...
}

理论上，如果您在第二层有另一个带有coord元素的架构，它将获得相同的全名，但由于这些不是架构注册表中需要引用的单个条目，因此不会导致冲突。无法从JSON Schema控制AVRO记录的命名空间有点遗憾，因为感觉就快到了，但我还没有能够深入挖掘以提供解决方案。

建议的SMT SetSchemaMetadata（参见问题的第一个答案）在这个过程中可能很有用，但它的文档与AVRO命名约定有点冲突，因为它在一个示例中显示了顺序值。它将试图找到一个架构，其中包含一个以该名称作为根元素的AVRO记录，并且由于“-”在AVRO名称中是非法字符，因此会出现一个错误。但是，如果使用了根元素的正确名称，SMT会做一些非常有用的事情：它的RestService类查询模式注册表以找到匹配的模式，失败时会显示一条消息，打印出需要创建的确切模式定义，因此不必记住所有转换规则。

因此，原始问题的答案是：是的，可以通过Kafka连接实现。如果你愿意，这也是最好的方式

• 不想编写自己的制作人/连接器

如果在数据摄取后进行转换是一种选择，那么ksqlDB的de、re和序列化功能似乎相当强大。