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] 错误 {org.wso2.extension.siddhi.map.json.sourcemapper.JsonSourceMapper} - 流“myDataStream”没有名为“ABC”的属性，但接收到的事件 {” event":{"ABC":"1"}} 确实如此。因此丢弃该消息。检查 json 字符串的默认映射格式是否正确。

我试过添加属性

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

语法错误：

在流“myDataStream”中定义的“json”错误，未映射属性“json”

任何帮助将不胜感激。

Answer 1

流的语法有错误，

define stream myDataStream (ABC string);

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