當一段時間內沒有收到元素時，我正在使用 flink Session 視窗，即；當發生不活動間隙時，它應該發出一個事件。
我在 flink 作業中將間隙配置為 10 秒。我發送了 event1 并在 5 秒后發送了 event2。這兩個事件應該屬于第一個視窗。輸出應該是這兩個事件的聚合。但我只得到第一個事件。
下面是我試過的代碼：
fun setupJob(env: StreamExecutionEnvironment) {
val testStream = env.sampleStream()
.keyBy { it.f0 }
.window(EventTimeSessionWindows.withGap(Time.seconds(10)))
.process(MyProcessWindowFunction())
testStream.map { it.toKafkaMessage() }
.kafkaSink<SampleOutput>() }
}
然后 MyProcessWindowFunction 看起來像
class MyProcessWindowFunction : ProcessWindowFunction<Tuple4<String, inputA?, inputB?, inputC?>, Tuple2<String, SampleOutput?>,
String, TimeWindow>() {
private lateinit var sampleOutputState: ValueState<SampleOutputState>
override fun open(parameters: Configuration) {
val SampleOutputStateDescriptor = ValueStateDescriptor("sample-output-state", SampleOutputState::class.java)
SampleOutputState = runtimeContext.getState(SampleOutputStateDescriptor)
}
override fun process(key: String, context: Context, elements: MutableIterable<Tuple4<String, inputA?, inputB?, inputC?>, out: Collector<Tuple2<String, SampleOutput?>>) {
val current = sampleOutputState.value()
val value = elements.iterator().next()
val latestState = when {
value.f2 != null -> processCondition(value.f2!!, current)
else -> return
}
sampleOutputState.update(latestState)
out.collect(Tuple2(key, latestState))
}
private fun processInputB(inputB: InputB, currentState: SampleOutputState?): SampleOutputState {
return currentState?.copy(
timestamp = System.currentTimeMillis(),
eventTime = condition.eventTime,
) ?:
createInputBState(inputB)
}
private fun createInputBState(inputB: InputB): SampleOutputState = SampleOutputState(
id = UUID.randomUUID().toString(),
timestamp = System.currentTimeMillis(),
eventTime = condition.eventTime,
)
}
我得到了唯一的 event1，但我想獲得這兩個事件的聚合（我發送了 event1 和 event2）。
我們如何獲得會話中可用事件的聚合？
uj5u.com熱心網友回復：
分配給視窗的所有事件都將在可迭代中發送到process您的ProcessWindowFunction. 您目前只查看第一個元素
val value = elements.iterator().next()
您需要迭代elements以產生聚合結果。
轉載請註明出處，本文鏈接：https://www.uj5u.com/net/436537.html