使用KStream语义重新组合

Question

使用 kafka 流，我想按某个键 K1 对元素 E 的流 S 进行分组，同时将同一键的所有值聚合到一个连接的结果 AGG 中。这会产生 KTable T1。

根据聚合结果，该值应重新分区到另一个 KTable T2 中，该 T2 按从聚合结果 AGG 中获取的键 K2 分组。因此，聚合结果应为下一次重新分组生成密钥。

最后，我只对一个KTableT2感兴趣，其中键是K2，值是AGG

但是，这不起作用。我只得到最后一个值的 KTable T。不是每个键 K2 的值

我知道聚合结果只会在一段时间后转发，所以我已经尝试降低commit.interval。ms设置为1，但无效。

我还尝试使用< code>through并将中间结果写入一个流中，但也没有成功。

val finalTable = streamBuilder.kstream("streamS")
                    .groupBy{ k, v -> createKey1(k, v) }
                    .aggregate(
                            { Agg.empty() },
                            { k, v, previousAgg ->
                                Agg.merge(previousAgg, v)
                            })
                    .toStream()
//                    .through("table1")
                    .groupBy { k1, agg -> agg.createKey2()}
                    .reduce{ _, agg -> agg }

对于包含以下值的流S：
key1=“123”，id=“1”，startNewGroup=“false”
key1=“234”，id=“2”，startNew Group=“false”
>key1=”123“，id=”3“，startNewGroup=”false“
key1=”123“”，id=”4“，startNew Group=”true“
>key1=”234“，id=“5”，startNewGroup=“false”
key1=“123”，id=“6”，startNewGroup=“false”
key1=“123”，id=“7”，startNewGroup:“false”
key1=“23”，id=“8”，startNew Group=“true”

我希望最终结果是一个具有以下最新键值的Ktable：
key： 123-1，value：'key 1="123"，key 2="123-1"，id="1,3"
key： 234-2，value：'key 1="234"，key 2="234-2"，id="2,5"
key： 123-4，value：'key 1="123"，key 2="123-4"，id="4,6,7"'
key： 123-8，value：'key 1="123"，key 2="123-8"，id="8"'

元素的原始流S首先由key1进行分组，其中聚合结果包含groupby keykey 1，并添加一个额外的字段key2（包含key 1与第一次出现的id的组合）
每当聚合收到startNewGroup设置为true的值时，它都会返回一个聚合，其中key2字段设置为新值的key1和id，从而有效地创建了一个新的子组
在第二次重新分组中，我们通过key2字段进行简单分组。

然而，我们真正观察到的是：
键：234-2，值：'key1=“234”，key2=“234-2”，ids=“2,5”'
键=123-8，值：'key1=“123”，key1=“123-8”，id=“8”'

Answer 1

对于您的用例，最好使用处理器 API。处理器API可以很容易地与Kafka Streams DSL（处理器API集成）结合使用。

您必须实现CustomTransformer，它将为状态存储中的特定密钥聚合消息。当＜code＞startNewGroup＝true＜/code＞消息到达时，密钥的旧消息将转发到下游，新聚合将开始

您示例转换器可能如下所示:

import org.apache.kafka.streams.kstream.Transformer
import org.apache.kafka.streams.processor.ProcessorContext
import org.apache.kafka.streams.state.KeyValueStore

case class CustomTransformer(storeName: String) extends Transformer[String, Value, Agg] {

  private var stateStore: KeyValueStore[String, Agg] = null
  private var context: ProcessorContext = null

  override def init(context: ProcessorContext): Unit = {
    this.context = context
    stateStore = context.getStateStore(storeName).asInstanceOf[KeyValueStore[String, Agg]]
  }

  override def transform(key: String, value: Value): Agg = {
    val maybeAgg = Option(stateStore.get(key))

    if (value.startNewGroup) {
      maybeAgg.foreach(context.forward(key, _))
      stateStore.put(key, Agg(value))
    }
    else
      stateStore.put(key, maybeAgg.map(_.merge(value)).getOrElse(Agg(value)))
    null
  }

  override def close(): Unit = {}
}