apache-kafka - WSO2 SP - 具有 JSON 属性的 Kafka 源

Question

我正在尝试使用以下代码从 Kafka 读取 JSON 数据:

@source(type = 'kafka', bootstrap.servers = 'localhost:9092', topic.list = 'TestTopic', 
group.id = 'test', threading.option = 'single.thread', @map(type = 'json'))

define stream myDataStream (json object);

但因以下错误而失败:

[2019-03-27_11-39-32_103] ERROR {org.wso2.extension.siddhi.map.json.sourcemapper.JsonSourceMapper} - Stream "myDataStream" does not have an attribute named "ABC", but the received event {"event":{"ABC":"1"}} does. Hence dropping the message. Check whether the json string is in a correct format for default mapping.

我试过添加属性

@source(type = 'kafka', bootstrap.servers = 'localhost:9092', 
topic.list = 'TestTopic', group.id = 'test', 
threading.option = 'single.thread', 
@map(type = 'json', @attributes(ABC = '$.ABC')))

语法错误:

Error at 'json' defined at stream 'myDataStream', attribute 'json' is not mapped

任何帮助将不胜感激。

Answer 1

流的语法有错误，

define stream myDataStream (ABC string);

这里的属性名称是 JSON 消息的键，在本例中为 ABC