在配对中找不到较早的匹配事件时Flink发出事件


0

我有两个事件流:一个是发出一个事件来标志一个项目生命周期的开始,另一个是发出一个事件来标志一个项目生命周期的结束。(流可以在itemId上 join。)

如何在Flink中为每个只有“生命周期结束”事件而没有相应开始的itemId1发出一个新事件?(这些开始和结束的事件可能相隔数小时或几天。)

1 答案


0

您可以在KeyedStream上使用有状态FlatMapFunction来实现该功能。

下面的代码片段应该能很好地满足您的需要。

val stream1: DataStream[Event1] = ???
val stream2: DataStream[Event2] = ???

// map both streams to their ID and a isStart flag to have a common type
val ids1: DataStream[(Int, Boolean)] = stream1.map(e => (e.id, true) )
val ids2: DataStream[(Int, Boolean)] = stream2.map(e => (e.id, false) )

// union both streams
val ids = ids1.union(ids2)

// use a stateful FlatMapFunction to check
val onlyEOL: DataStream[Int] = ids
// organize stream by ID
.keyBy(.1)
// use stateful FlatMapFunction to check that bol arrived before eol
.flatMapWithState {
(value: (Int, Boolean), state: Option[Boolean]) =>
if (value._2) {
// bol event -> emit nothing and set state to true
( List(), Some(true))
} else {
// eol event
if (state.isDefined && state.get) {
// bol was seen before -> emit nothing and remove state
( List(), None)
} else {
// bol was NOT seen before -> emit ID and remove state
( List(value._1), None)
}
}
}


我来回答

写文章

提问题

面试题