SlidingEventTimeWindows不产生任何输出

Question

我有一个配置为的流执行

StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
DataStreamSource<Record> stream = env.addSource(new FlinkKafkaConsumer(
SystemsCpu.TOPIC,
ConfluentRegistryAvroDeserializationSchema.forGeneric(SystemsCpu.SCHEMA, registry),
config)
.setStartFromLatest());
DataStream<Anomaly> anomalies = stream
.keyBy(x -> x.get("host").toString())
.window(SlidingEventTimeWindows.of(Time.minutes(20), Time.seconds(20))) // produces output with TumblingEventTimeWindows
.process(new AnomalyDetector())
.name("anomaly-detector");

public class AnomalyDetector extends ProcessWindowFunction<Record, Anomaly, String, TimeWindow> {
@Override
public void process(String key, Context context, Iterable<Record> input, Collector<Anomaly> out) {
var anomaly = new Anomaly();
anomaly.setValue(1.0);
out.collect(anomaly);
}
}

然而，由于某种原因，SlidingEventTimeWindows不产生任何要由AnomalyDetector处理的输出(即，过程根本不被触发(。例如，如果我使用TumblingEventTimeWindows，它会按预期工作。

你知道是什么原因造成的吗？我是否错误地使用了SlidingEventTimeWindows？

Answer 1

在执行任何类型的事件时间窗口时，都需要提供WatermarkStrategy。水印标记流中的一个点，并表示流在某个特定时间点完成。事件时间窗口只能由足够大的水印的到达来触发。

有关详细信息，请参阅文档，但可能是这样的：

DataStream<MyType> timestampedEvents = events
.assignTimestampsAndWatermarks(
WatermarkStrategy
.<MyType>forBoundedOutOfOrderness(Duration.ofSeconds(10))
.withTimestampAssigner((event, timestamp) -> event.timestamp));

然而，由于您使用的是Kafka，通常最好让Flink Kafka消费者进行水印：

FlinkKafkaConsumer<MyType> kafkaSource = new FlinkKafkaConsumer<>("myTopic", schema, props);
kafkaSource.assignTimestampsAndWatermarks(WatermarkStrategy...);
DataStream<MyType> stream = env.addSource(kafkaSource);

请注意，如果您使用这种稍后的方法，并且您的事件在每个Kafka分区内按时间顺序排列，则可以利用Flink Kafka源提供的每分区水印，并使用WatermarkStrategy.forMonotonousTimestamps()而不是有界有序策略。这有很多优点。

顺便说一句，这与您的问题无关，但您应该注意，通过指定SlidingEventTimeWindows.of(Time.minutes(20), Time.seconds(20))，每个事件都将复制到60个重叠窗口中的每个窗口中。

Answer 2

您使用的是SlidingEventTimeWindows，但您的流执行环境默认配置为处理时间。使用SlidingProcessingTimeWindows或为事件时间配置您的环境，如

env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)

活动时间还需要一个特殊的时间戳分配器，您可以在这里找到更多信息。

https://www.ververica.com/blog/stream-processing-introduction-event-time-apache-flink?hs_amp=true