0

我目前有一个流，具有 1 个包含 Json 事件的字符串属性。

这个流接收不同的事件，我想应用 Json 路径表达式，以便我可以在过滤器和函数上使用这些属性。

JsonPath 提取器在过滤器和选择器上就像一个魅力，不幸的是，我无法将它们用于“分组依据”部分。我实际上是在手动添加了 siddhi-execution-json 扩展的嵌入式 Siddhi 应用程序中执行此操作，但为了讨论，以便每个人都可以轻松检查和测试它，我将粘贴一个适用于 WSO2 流处理器的示例应用程序。目标看起来像以下应用程序：

@App:name("Group_by_json_attribute")

define stream JsonStream(json string);

@sink(type='log')
define stream LogStream(myField string, count long);

@info(name='query1')
from JsonStream#window.time(10 sec)
select json:getString(json, '$.myField') as myField, count() as count 
group by myField having count > 1 
insert into LogStream;

它可以接受以下事件：

{"myField": "my_value"}

但是，此查询将引发错误：

Cannot find attribute type as 'myField' does not exist in 'JsonStream'; define stream JsonStream(json string)

我还尝试在“Group by”中直接使用 Json 提取器：

group by json:getString(json, '$.myField') as myField having count > 1

但是现在的错误是：

mismatched input ':' expecting {',', ORDER, LIMIT, OFFSET, HAVING, INSERT, DELETE, UPDATE, RETURN, OUTPUT}

这似乎不希望在这里使用扩展

我只是想知道，是否可以按输入流中未直接定义的属性进行分组。在这种情况下，是从 JSON 对象中提取的字段，但它可以是生成另一个属性的任何其他函数。

我也在使用来自 Maven 中央存储库的版本

  • 悉地语： io.siddhi :siddhi-core:5.0.1
  • siddhi-execution-json：io.siddhi.extension.execution.json:siddhi-execution-json:2.0.1

（编辑）澄清

目标是使用未直接在 Stream 中定义的属性，用于Group By.

原因是，我目前有一个嵌入式应用程序，它定义了来自外部源的整套输入流，格式为 JSON，还有一组输出流在查询匹配时通知外部组件。这个应用程序允许用户在这组预定义的 Streams 上创建自定义查询，但他们不能自己创建 Streams。

非常感谢！

4

1 回答 1

0

似乎我们期望查询输入流中的按字段分组，在本例中为 JsonStream。在此之前使用另一个查询进行提取以及以下查询中的聚合和过滤，

@App:name("Group_by_json_attribute")

define stream JsonStream(json string);

@sink(type='log')
define stream LogStream(myField string, count long);

@info(name='extract_stream')
from JsonStream
select json:getString(json, '$.myField') as myField 
insert into ExtractedStream;

@info(name='query1')
from ExtractedStream#window.time(10 sec)
select myField, count() as count 
group by myField 
having count > 1 
insert into LogStream;
于 2019-07-03T05:15:23.753 回答