如何解决无法使用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
并没有被其他代码覆盖。
版权声明:本文内容由互联网用户自发贡献,该文观点与技术仅代表作者本人。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如发现本站有涉嫌侵权/违法违规的内容, 请发送邮件至 dio@foxmail.com 举报,一经查实,本站将立刻删除。