3

我的问题是关于 Kafka 流 Ktable.groupBy.aggregate。以及由此产生的聚合值。

情况

我正在尝试每天汇总分钟事件。

我有一个分钟事件生成器（此处未显示），可为一些房屋生成事件。有时事件值错误，必须重新发布分钟事件。 分钟事件发布在“分钟”主题中。

我正在使用kafka StreamsgroupBy和aggregate.

问题

通常，由于一天有 1440 分钟，因此不应有超过 1440 个值的聚合。此外，永远不应该有一个包含负数事件的聚合。

...但无论如何它都会发生，我们不明白我们的代码有什么问题。

示例代码

这是一个示例简化代码来说明问题。有时会抛出 IllegalStateException。


        StreamsBuilder builder = new StreamsBuilder();

        KTable<String, MinuteEvent> minuteEvents = builder.table(
                "minutes",
                Consumed.with(Serdes.String(), minuteEventSerdes),
                Materialized.<String, MinuteEvent, KeyValueStore<Bytes, byte[]>>with(Serdes.String(), minuteEventSerdes)
                        .withCachingDisabled());

        // preform daily aggregation
        KStream<String, MinuteAggregate> dayEvents = minuteEvents
                // group by house and day
                .filter((key, minuteEvent) -> minuteEvent != null && StringUtils.isNotBlank(minuteEvent.house))
                .groupBy((key, minuteEvent) -> KeyValue.pair(
                        minuteEvent.house + "##" + minuteEvent.instant.atZone(ZoneId.of("Europe/Paris")).truncatedTo(ChronoUnit.DAYS), minuteEvent),
                        Grouped.<String, MinuteEvent>as("minuteEventsPerHouse")
                                .withKeySerde(Serdes.String())
                                .withValueSerde(minuteEventSerdes))
                .aggregate(
                        MinuteAggregate::new,
                        (String key, MinuteEvent value, MinuteAggregate aggregate) -> aggregate.addLine(key, value),
                        (String key, MinuteEvent value, MinuteAggregate aggregate) -> aggregate.removeLine(key, value),
                        Materialized
                                .<String, MinuteAggregate, KeyValueStore<Bytes, byte[]>>as(BILLLINEMINUTEAGG_STORE)
                                .withKeySerde(Serdes.String())
                                .withValueSerde(minuteAggSerdes)
                                .withLoggingEnabled(new HashMap<>())) // keep this aggregate state forever
                .toStream();

        // check daily aggregation
        dayEvents.filter((key, value) -> {
            if (value.nbValues < 0) {
                throw new IllegalStateException("got an aggregate with a negative number of values " + value.nbValues);
            }
            if (value.nbValues > 1440) {
                throw new IllegalStateException("got an aggregate with too many values " + value.nbValues);
            }
            return true;
        }).to("days", minuteAggSerdes);

以下是此代码片段中使用的示例类：

    public class MinuteEvent {
        public final String house;
        public final double sensorValue;
        public final Instant instant;

        public MinuteEvent(String house,double sensorValue, Instant instant) {
            this.house = house;
            this.sensorValue = sensorValue;
            this.instant = instant;
        }
    }

    public class MinuteAggregate {
        public int nbValues = 0;
        public double totalSensorValue = 0.;
        public String house = "";

        public MinuteAggregate addLine(String key, MinuteEvent value) {
            this.nbValues = this.nbValues + 1;
            this.totalSensorValue = this.totalSensorValue + value.sensorValue;
            this.house = value.house;
            return this;
        }

        public MinuteAggregate removeLine(String key, MinuteEvent value) {
            this.nbValues = this.nbValues -1;
            this.totalSensorValue = this.totalSensorValue - value.sensorValue;
            return this;
        }

        public MinuteAggregate() {
        }
    }

如果有人能告诉我们我们在这里做错了什么以及为什么我们有这些意想不到的价值，那就太好了。

补充说明

  • 我们将流作业配置为使用 4 个线程运行properties.put(StreamsConfig.NUM_STREAM_THREADS_CONFIG, 4);
  • 我们被迫使用 a ，因为对于已发布的 Instant，可以使用不同的Ktable.groupBy().aggregate()分钟值 重新发布。sensorValue并且每日聚合相应地修改。 Stream.groupBy().aggregate()没有adderAND substractor。
4

1 回答 1

1

我认为，计数实际上有可能暂时变为负数。

原因是，您的第一个更新中的每个更新都会向下游KTable发送两个消息——要在下游聚合中减去旧值，以及要添加到下游聚合中的新值。两条消息都将在下游聚合中独立处理。

如果当前计数为零，并且在加法之前处理了减法，则计数将暂时变为负数。

于 2019-10-20T00:58:41.770 回答