0

当一段时间内没有收到元素时，我正在使用 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）。

我们如何获得会话中可用事件的聚合？

4

1 回答 1

0

分配给窗口的所有事件都将在可迭代中发送到process您的ProcessWindowFunction. 您目前只查看第一个元素

val value = elements.iterator().next()

您需要迭代elements以产生聚合结果。

于 2022-02-23T09:01:15.597 回答