如何在流数据的时间戳基础上处理FLINK窗口？

Question

我有一些问题。

基于类中的时间戳，我想做一个逻辑，排除在1分钟内输入N次或更多次的数据。

UserData类有一个时间戳变量。

class UserData{ 
      public Timestamp timestamp; 
      public String userId; 
    }

起初我试着用一个翻滚的窗户。

SingleOutputStreamOperator<UserData> validStream =
                stream.keyBy((KeySelector<UserData, String>) value -> value.userId)
                        .window(TumblingProcessingTimeWindows.of(Time.seconds(60)))
                        .process(new ValidProcessWindow());

public class ValidProcessWindow extends ProcessWindowFunction<UserData, UserData, String, TimeWindow> {

    private int validCount = 10;

    @Override
    public void process(String key, Context context, Iterable<UserData> elements, Collector<UserData> out) throws Exception {
        int count = -1;
        for (UserData element : elements) {
            count++; // start is 0

            if (count >= validCount) // valid click count
            {
                continue;
            }
            
            out.collect(element);
        }
    }
}

但是，滚动窗口的时间计算是基于固定时间的，因此无论UserData类的时间戳如何，它都不适合。

如何处理流上窗口UserData类的时间戳基？

谢谢。

附加信息

我使用这样的代码。

stream.assignTimestampsAndWatermarks(WatermarkStrategy.<UserData>forBoundedOutOfOrderness(Duration.ofSeconds(1))                
.withTimestampAssigner((event, timestamp) -> Timestamps.toMillis(event.timestamp))
.keyBy((KeySelector<UserData, String>) value -> value.userId)
.window(TumblingEventTimeWindows.of(Time.seconds(60)))
.process(new ValidProcessWindow());

我试了一些测试。150个样本数据。每个数据的时间戳增加1秒。结果是|1,2,3....59||60,61....119|。我等待最后30个数据。我期待|1,2,3....59||60,61....119||120...149|。

如何获得最后的其他数据？

自我回答

我找到了原因。因为我只使用了150个样本数据。

如果使用事件时间在Flink不能进展，如果没有要处理的元素。

https://ci.apache.org/projects/flink/flink-docs-release-1.9/dev/event_time.html#idling-sources

因此，我测试了150个样本数据和虚拟数据。（每个数据的虚拟数据时间戳增加1秒）。

我收到了正确的数据| 1,2,3。。。。59| |60,61....119| |120...149|.

谢谢你的帮助。

Answer 1

所以就我理解你的问题，你应该使用不同的时间特征。处理时间是使用系统时间来计算窗口，您应该为您的应用程序使用事件时间。您可以在这里找到关于正确使用活动时间的更多信息。

编辑：这就是flink的工作原理，没有数据将水印推到150以上，所以窗口没有关闭，因此没有输出。您可以使用自定义触发器，即使没有生成水印也会关闭窗口，或者注入一些数据来移动水印。