Kafka 流处理批处理数据以重置聚合

Question

我有一些数据到达我的 kafka 主题“数据源”，具有以下模式（此处为演示进行了简化）：

{ "deal" : -1, "location": "", "value": -1, "type": "init" }
{ "deal": 123456, "location": "Mars", "value": 100.0, "type": "batch" },
{ "deal" 123457, "location": "Earth", "value", 200.0, "type": "batch" },
{ "deal": -1, "location": "", "value", -1, "type": "commit" }

此数据来自批次运行，我们获取所有交易并重新计算它们的价值。将其视为一天开始的过程 - 此时，这是所有位置的一组新数据。 目前init和commit消息没有发送到真正的主题，它们被生产者过滤掉了。

白天，随着事情的变化会有更新。这提供了新数据（在这个例子中我们可以忽略覆盖数据，因为这将通过重新运行批处理来处理）：

{ "deal": 123458, "location", "Mars", "value": 150.0, "type": "update" }

此数据作为 KStream“位置”进入应用程序。

另一个主题“位置”列出了可能的位置。这些被拉入 java kafka-streams 应用程序作为 KGlobalTable 位置：

{ "id": 1, "name": "Mars" },
{ "id": 2, "name": "Earth"}

计划使用 java 9 kafka-streams 应用程序来聚合这些按位置分组的值。输出应该类似于：

{ "id": 1, "location": "Earth", "sum": 250.0 },
{ "id": 2, "location": "Mars": "sum": 200.0 }

这是我目前所做的工作：

StreamsBuilder builder = new StreamsBuilder();

/** snip creating serdes, settings up stores, boilerplate  **/

final GlobalKTable<Integer, Location> locations = builder.globalTable(
                LOCATIONS_TOPIC, 
                /* serdes, materialized, etc */
                );

final KStream<Integer, PositionValue> positions = builder.stream(
                POSITIONS_TOPIC,
                /* serdes, materialized, etc */
            );

/* The real thing is more than just a name, so a transformer is used to match locations to position values, and filter ones that we don't care about */
KStream<Location, PositionValue> joined = positions
                .transform(() -> new LocationTransformer(), POSITION_STORE) 
                .peek((location, positionValue) -> { 
                    LOG.debugv("Processed position {0} against location {1}", positionValue, location);
                });

/** This is where it is grouped and aggregated here **/
joined.groupByKey(Grouped.with(locationSerde, positionValueSerde))
            .aggregate(Aggregation::new, /* initializer */
                       (location, positionValue, aggregation) -> aggregation.updateFrom(location, positionValue), /* adder */
                Materialized.<Location, Aggregation>as(aggrStoreSupplier)
                    .withKeySerde(locationSerde)
                    .withValueSerde(aggregationSerde)
            );

Topology topo = builder.build();

我遇到的问题是，这是在聚合所有内容 - 所以每天的批次、加上更新，然后是下一个每天的批次，都会被添加。基本上，我需要一种方式来表达“这是下一组批处理数据，针对此进行重置”。我不知道该怎么做 - 请帮忙！

谢谢

Answer 1

所以如果我没理解错的话，你想要汇总数据，但仅限于最后一天，并丢弃其余数据。

我建议您聚合到一个中介 class 中，它包含流中的所有值，并且还具有过滤掉其他日子数据的逻辑。如果我理解正确的话，那就是丢弃最后一个“批处理”类型之前的所有数据。

虽然在Kotlin中，我做了一个similar solution，有需要的可以看看

Answer 2

您可以做一些事情，但我建议您使用 TimeWindowed Stream。您可以将时间设置为 1 天的滚动 window 并对该流执行仲裁。您最终会在 KTable 中汇总到自己的 window 中。然后你就不用担心丢弃数据（虽然你可以）并且每天都会分开。

这里有几个很好的例子说明它们是如何工作的：https://www.programcreek.com/java-api-examples/?api=org.apache.kafka.streams.kstream.TimeWindows

Kafka 流处理批处理数据以重置聚合

Kafka streams handle batch data to reset aggregation

java

apache-kafka

apache-kafka-streams