WSO2 Streaming Integrator - Siddhi.io 将多个事件聚合为一个事件

Question

我需要将多个事件收集到一个事件中，并将其视为一个事件。输出流应包括输入事件列表。

基本上，当以下事件通过时

{InvoiceNumber:1, LineItem: 1, Value: 100}
{InvoiceNumber:1, LineItem: 2, Value: 200}
{InvoiceNumber:1, LineItem: 3, Value: 300}

我需要输出如下所示

[InvoiceNumber:1, lineItems: [{LineItem: 1, Value: 100}, {LineItem: 2, Value: 200}, {LineItem: 3, Value: 300}]

如何使用 WSO2 Streaming Integrator 实现这一点？或者 Siddhi.io.

我尝试了以下操作，但它仍然将每个流插入到输出流中

partition with (InvoiceNo of CSVInputStream)
begin

    from every e1=CSVInputStream
        within 1 min
    select e1.InvoiceNo, list:collect(e1.LineItem) as lineItems
    group by e1.InvoiceNo
    insert into AggregatedStream;
end;

Answer 1

不要使用分区，因为这是一个简单的用例，请尝试 windows。在您的情况下，时间批次 windows，https://siddhi.io/en/v5.1/docs/api/latest/#timebatch-window

from CSVInputStream#window.timeBatch(1 min)
select e1.InvoiceNo, list:collect(e1.LineItem) as lineItems
group by e1.InvoiceNo
insert into AggregatedStream;

WSO2 Streaming Integrator - Siddhi.io 将多个事件聚合为一个事件

WSO2 Streaming Integrator - Siddhi.io Aggregate multiple events into one event

wso2

siddhi

wso2si