无法使用python将avro数据写入kafka

问题描述

我正在使用kafka kafka_2.11-0.11.0.2和融合版本3.3.0进行架构注册。

我已经定义了如下的avro模式：

{
"namespace": "com.myntra.search","type": "record","name": "SearchDataIngestionObject","fields": [
  {"name": "timestamp","type":"long"},{"name": "brandList","type":{ "type" : "array","items" : "string" }},{"name": "articleTypeList",{"name": "gender",{"name": "masterCategoryList",{"name": "subCategoryList",{"name": "quAlgo",{"name": "colours",{"name": "isLandingPage","type": "boolean"},{"name": "isUserQuery",{"name": "isAutoSuggest",{"name": "userQuery","type": "string"},{"name": "correctedQuery",{"name": "completeSolrQuery",{"name": "atsaList","type":{"type": "map","values":{ "type" : "array","items" : "string" }}},{"name": "quMeta","type": {"type": "map","values": "string"}},{"name": "requestId","type": "string"}
]

}

我正尝试将一些数据写入kafka，如下所示：

value = {
    "timestamp": 1597399323000,"brandList": ["brand_value"],"articleTypeList": ["articleType_value"],"gender": ["gender_value"],"masterCategoryList": ["masterCategory_value"],"subCategoryList": ["subCategory_value"],"quAlgo": ["quAlgo_value"],"colours": ["colours_value"],"isLandingPage": False,"isUserQuery": False,"isAutoSuggest": False,"userQuery": "userQuery_value","correctedQuery": "correctedQuery_value","completeSolrQuery": "completeSolrQuery_value","atsaList": {
        "atsa_key1": ["atsa_value1"],"atsa_key2": ["atsa_value2"],"atsa_key3": ["atsa_value3"]
    },"quMeta": {
        "quMeta_key1": "quMeta_value1","quMeta_key2": "quMeta_value2","quMeta_key3": "quMeta_value3"
    },"requestId": "requestId_value"
}

topic = "search"
key = str(uuid.uuid4())

producer.produce(topic=topic,key=key,value=value)
producer.flush()

但是我遇到了以下错误：

Traceback (most recent call last):
File "producer.py",line 61,in <module>
  producer.produce(topic=topic,value=value)
File "/Library/Python/2.7/site-packages/confluent_kafka/avro/__init__.py",line 99,in produce
  value = self._serializer.encode_record_with_schema(topic,value_schema,value)
File "/Library/Python/2.7/site-packages/confluent_kafka/avro/serializer/message_serializer.py",line 118,in encode_record_with_schema
  return self.encode_record_with_schema_id(schema_id,record,is_key=is_key)
File "/Library/Python/2.7/site-packages/confluent_kafka/avro/serializer/message_serializer.py",line 152,in encode_record_with_schema_id
  writer(record,outf)
File "/Library/Python/2.7/site-packages/confluent_kafka/avro/serializer/message_serializer.py",line 86,in <lambda>
  return lambda record,fp: writer.write(record,avro.io.BinaryEncoder(fp))
File "/Library/Python/2.7/site-packages/avro/io.py",line 979,in write
  raise AvroTypeException(self.writers_schema,datum)
avro.io.AvroTypeException: The datum {'quAlgo': ['quAlgo_value'],'userQuery': 'userQuery_value','isAutoSuggest': False,'isLandingPage': False,'timestamp': 1597399323000,'articleTypeList': ['articleType_value'],'colours': ['colours_value'],'correctedQuery': 'correctedQuery_value','quMeta': {'quMeta_key1': 'quMeta_value1','quMeta_key2': 'quMeta_value2','quMeta_key3': 'quMeta_value3'},'requestId': 'requestId_value','gender': ['gender_value'],'isUserQuery': False,'brandList': ['brand_value'],'masterCategoryList': ['masterCategory_value'],'subCategoryList': ['subCategory_value'],'completeSolrQuery': 'completeSolrQuery_value','atsaList': {'atsa_key1': ['atsa_value1'],'atsa_key2': ['atsa_value2'],'atsa_key3': ['atsa_value3']}} is not an example of the schema {
"namespace": "com.myntra.search","fields": [
  {
    "type": "long","name": "timestamp"
  },{
    "type": {
      "items": "string","type": "array"
    },"name": "brandList"
  },"name": "articleTypeList"
  },"name": "gender"
  },"name": "masterCategoryList"
  },"name": "subCategoryList"
  },"name": "quAlgo"
  },"name": "colours"
  },{
    "type": "boolean","name": "isLandingPage"
  },"name": "isUserQuery"
  },"name": "isAutoSuggest"
  },{
    "type": "string","name": "userQuery"
  },"name": "correctedQuery"
  },"name": "completeSolrQuery"
  },{
    "type": {
      "values": {
        "items": "string","type": "array"
      },"type": "map"
    },"name": "atsaList"
  },{
    "type": {
      "values": "string","name": "quMeta"
  },"name": "requestId"
  }
]
}

我什至尝试使用与给定的here相同的示例，但是它不起作用并抛出相同的错误。

解决方法

在您的例外情况下，错误提示您所提供的数据如下：

{'userQuery': 'userQuery_value','isAutoSuggest': False,'isLandingPage': False,'correctedQuery': 'correctedQuery_value','isUserQuery': False,'timestamp': 1597399323000,'completeSolrQuery': 'completeSolrQuery_value','requestId': 'requestId_value'}

这远低于您在示例中提供的声称。

在执行producer.produce(topic=topic,key=key,value=value)之前，您可以返回原始代码并在第60行进行简单的print(value)来确保您发送的是正确的值，并且value并没有被其他代码覆盖。