Flink-在时间限制内处理连续事件

Question

我有一个用例，我想我需要一些关于如何处理它的帮助。因为我是流媒体和Flink的新手，所以我会尽量描述我想要实现的目标。对不起，如果我没有使用正式和正确的语言。

我的代码将用java编写，但我不想用python编写代码，也不想用伪代码或方法编写代码。

TL： DR

1. 在某个时间限制内对同一密钥的事件进行分组

背景：

1. 我有来自传感器的数据，就像Kafka的数据流一样。
2. 我需要使用eventTime，因为该数据未记录。给我90%事件的迟到时间大约是1分钟。
3. 我把这些事件归类为一些关键字。

我想做的是：

1. 根据某些事件的字段，我想将2个事件“加入/混合”到一个新事件（“结果事件”）中
2. 第一个条件是，这些连续事件之间的间隔不超过30秒
3. 接下来的条件只是检查一些字段值，然后再决定

我的psuedo解决方案：

1. 为每个事件打开一个新窗口。那个窗口应该是1分钟。
2. 对于在那一分钟内发生的每一个事件-我想检查它的事件时间，看看它是否距离初始窗口事件30秒。如果是-检查其他条件并省略一个新的结果流。

问题-当新事件发生时，需要：

1. 为自己创建一个新窗口。
2. 在距离它30秒的几个可能窗口中只加入一个窗口。

问题是：

那有可能吗？

换句话说，我只在两个“连续”事件之间建立联系。

非常感谢你。

也许展示**BATCH案例的解决方案会显示我正在努力做到最好：**

for i in range(grouped_events.length):
    event_A = grouped_events[i]
    event_B = grouped_events[i+1]
    if event_B.get("time") - event_A.get("time") < 30:
        if event_B.get("color") == event_A.get("color"):
            if event_B.get("size") > event_A.get("size"):
                create_result_event(event_A, event_B)

我的（天真的）尝试到目前为止与Flink在java

**求和函数只是我的函数创建新结果对象的占位符。。。

>

第一种解决方案是只做一个简单的时间窗口，然后按某个字段求和

是否尝试在第二个窗口上运行一些事件并对所有事件进行迭代？

DataStream
.keyBy(threeEvent -> threeEvent.getUserId())
.window(TumblingEventTimeWindows.of(Time.seconds(60)))
.sum("size")
.print();



DataStream
.keyBy(threeEvent -> threeEvent.getUserId())
.window(TumblingEventTimeWindows.of(Time.seconds(5)))
.process(new processFunction());



public static class processFunction extends ProcessWindowFunction<ThreeEvent, Tuple3<Long, Long, Float>, Long, TimeWindow> {
    @Override
    public void process(Long key, Context context, Iterable<ThreeEvent> threeEvents, Collector<Tuple3<Long, Long, Float>> out) throws Exception {
        Float sumOfSize = 0F;
        for (ThreeEvent f : threeEvents) {
            sumOfSize += f.getSize();
        }

        out.collect(new Tuple3<>(context.window().getEnd(), key, sumOfTips));
    }
}

Answer 1

当然，您可以使用windows创建排序和分析的小批次，但很难正确处理窗口边界（如果应该配对的事件位于不同的窗口中会怎么样？）。

这看起来更容易通过键控流和有状态平面图来完成。只需使用RichFlatMapFunction，并使用一段键控状态（ValueState），它记住每个键的前一个事件。然后，在处理每个事件时，将其与保存的事件进行比较，如果发生这种情况，则生成结果，并更新状态。

您可以在flink训练和flink留档中阅读有关使用flink键控状态的内容。

关于你的用例，我关心的一件事是你的事件是否会乱序到达。是否需要首先按时间戳对事件进行排序才能获得正确的结果？这不是微不足道的。如果这是一个问题，那么我建议您使用FlinkSQLMATCH_RECOGNIZE或CEP库，这两个库都是为对事件流进行模式识别而设计的，并且会为您处理流的html" target="_blank">排序（您只需必须提供时间戳和水印）。

此查询可能并不完全正确，但希望能够传达如何使用match Recognite执行类似操作的风格：

SELECT * FROM Events
MATCH_RECOGNIZE (
  PARTITION BY userId
  ORDER BY eventTime
  MEASURES
    A.userId as userId,
    A.color as color,
    A.size as aSize,
    B.size as bSize
  AFTER MATCH SKIP PAST LAST ROW
  PATTERN (A B)
  DEFINE
    A AS true,
    B AS ( timestampDiff(SECOND, A.eventTime, B.eventTime) < 30) 
           AND A.color = B.color 
           AND A.size < B.size )
);

这也可以在CEP中非常自然地完成，其中比较连续事件的基础是使用迭代条件，您可以在子句中使用来处理时间约束。