用于嵌套记录的 Confluent Kafka 生产者消息格式答案

【问题标题】：Confluent Kafka producer message format for nested records用于嵌套记录的 Confluent Kafka 生产者消息格式
【发布时间】：2021-04-19 00:02:49
【问题描述】：

我在 kafka 主题中注册了一个 AVRO 模式，并正在尝试向它发送数据。架构有嵌套记录，我不确定如何使用 confluent_kafka python 正确地向它发送数据。

示例架构： *排除架构中的任何拼写错误（真实的非常大，只是一个示例）

 {
 "namespace": "company__name",
 "name": "our_data",
 "type": "record",
 "fields": [
           {
            "name": "datatype1",
            "type": ["null", {
                 "type": "record",
                 "name": "datatype1_1",
                 "fields": [ 
                     {"name": "site", "type": "string"},
                     {"name": "units", "type": "string"}
                  ]
             }]
             "default": null
            }
            {
            "name": "datatype2",
            "type": ["null", {
                 "type": "record",
                 "name": "datatype2_1",
                 "fields": [ 
                     {"name": "site", "type": "string"},
                     {"name": "units", "type": "string"}
                  ]
             }]
             "default": null
            }
           ]
          }

我正在尝试使用 confluent_kafka python 版本将数据发送到此架构。当我之前完成此操作时，记录没有嵌套，我将使用典型的字典 key: value 对并将其序列化。如何发送嵌套数据以使用架构。

到目前为止我尝试了什么......

message = {'datatype1': 
            {'site': 'sitename',
             'units': 'm'
            }
           }

这个版本不会导致任何kafka错误，但是所有的列都显示为null

还有……

message = {'datatype1': 
            {'datatype1_1':
              {'site': 'sitename',
               'units': 'm'
              }
            }
           }

此版本的架构产生了 kafka 错误。

【问题讨论】：

您在 Python 字典中缺少 datatype1 和 datatype1_1 的引号，因此这不是有效代码，但注册表或 Kafka 的错误是什么？
对不起，这只是写问题时的一个错字。刚刚解决了这个问题。 KafkaError{code=_VALUE_SERIALIZATION,val=-161,str="{'datatype1_1':{'site': 'sitename', 'units': 'm'}" (type ) 不匹配 (架构）
这是 JSON 还是二进制文件？
当datatype1_1 不可为空时，您是否尝试过它是否有效？

标签： apache-kafka avro confluent-platform confluent-schema-registry confluent-kafka-python

【解决方案1】：

如果您使用命名空间，则不必担心命名冲突，并且可以正确构建可选记录：例如，两者都

{
  "meta": {
    "instanceID": "something"
  }
}

和

{}

是以下的有效实例：

{
  "doc": "Survey",
  "name": "Survey",
  "type": "record",
  "fields": [
    {
      "name": "meta",
      "type": [
        "null",
        {
          "name": "meta",
          "type": "record",
          "fields": [
            {
              "name": "instanceID",
              "type": [
                "null",
                "string"
              ],
              "namespace": "Survey.meta"
            }
          ],
          "namespace": "Survey"
        }
      ],
      "namespace": "Survey"
    }
  ]
}

【讨论】：