WSO2 SP - 具有 JSON 属性的 Kafka 源答案

【问题标题】：WSO2 SP - Kafka source with JSON attributesWSO2 SP - 具有 JSON 属性的 Kafka 源
【发布时间】：2019-03-27 20:40:10
【问题描述】：

我正在尝试使用以下代码从 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”是未映射

任何帮助将不胜感激。

【问题讨论】：

标签： apache-kafka wso2 kafka-consumer-api kafka-producer-api wso2sp

【解决方案1】：

流的语法有错误，

define stream myDataStream (ABC string);

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

【讨论】：

非常感谢。我将有 100 个来自 kafka 的键/值对。那么我们是否必须将所有这些键映射到输入流？或者有没有办法将 JSON 作为对象访问并从中解析出所需的密钥？像 $.ABC 或 json.ABC 之类的东西？
是的，有可能，您可以使用自定义映射仅获取所需的键，@map(type = 'json', @attributes(key1 = '$.ABC.key')))定义流 myDataStream(key1 string);
非常感谢，这在很大程度上有所帮助。我的 JSON 将是 {"event":{"A":"1","B":"2","C":"3"}} 并添加属性为 @attributes(json = '$.event')。但是分配给变量的字符串不是 JSON 格式，它会转换 JSON key=value 格式，例如{A=1,B=2,C=3}。因此，我无法使用json:getString(json,"$.A") 进一步使用 JSON。
或者有没有办法在 wso2sp 或 siddhi 中将 {A=1,B=2,C=3} 转换回 {"A":"1","B":"2","C":"3"}？
如果我错了，请纠正我，本质上不是在源头映射属性，而是要在整个分析流程中保持 JSON 字符串不变？用于下游操作？