KTable值字段上的Kafka流分组

Question

我有一个用例，我的KTable是这样的。

KTable：orderTable

键：值

KTable:OrderByIdTable=>此表将位于groupBy值字段(id)上，且计数列值将具有ID1=(12+5)和ID2=(10+11)

键：值

{id1}：{17}

{id2}：{21}

         final KTable<String, Order> orderTable = builder.table("order-topic");
         Don't know how to do this further.....
         final KTable<String,Long> orderByIdTable = ?

Answer 1

下面是一个代码示例（仅使用Java基元类型，这使我更快地组合在一起），演示了如何对KTable进行重键和重分区，从而生成新的KTable。您应该能够很容易地使其适应将ktable 转换为ktable 的示例。

就我个人而言，我会为您的用例选择Variant 2。

下面的例子。没有经过充分测试，可能是tombstone记录（带有非null键但值为null的消息，表示应该从表中删除该键）没有得到正确处理。

final StreamsBuilder builder = new StreamsBuilder();
final KTable<Integer, String> table = builder.table(inputTopic, Consumed.with(Serdes.Integer(), Serdes.String()));

// Variant 1 (https://docs.confluent.io/current/streams/faq.html#option-1-write-kstream-to-ak-read-back-as-ktable)
// Here, we re-key the KTable, write the results to a new topic, and then re-read that topic into a new KTable.
table
    .toStream()
    .map((key, value) -> KeyValue.pair(value, key))
    .to(outputTopic1, Produced.with(Serdes.String(), Serdes.Integer()));
KTable<String, Integer> rekeyedTable1 =
    builder.table(outputTopic1, Consumed.with(Serdes.String(), Serdes.Integer()));

// Variant 2 (https://docs.confluent.io/current/streams/faq.html#option-2-perform-a-dummy-aggregation)
// Here, we re-key the KTable (resulting in a KGroupedTable), and then perform a dummy aggregation to turn the
// KGroupedTable into a KTable.
final KTable<String, Integer> rekeyedTable2 =
    table
        .groupBy(
            (key, value) -> KeyValue.pair(value, key),
            Grouped.with(Serdes.String(), Serdes.Integer())
        )
        // Dummy aggregation
        .reduce(
            (aggValue, newValue) -> newValue, /* adder */
            (aggValue, oldValue) -> oldValue  /* subtractor */
        );
rekeyedTable2.toStream().to(outputTopic2, Produced.with(Serdes.String(), Serdes.Integer()));