kafka streams groupBy聚合生成意外值

Question

我的问题是关于Kafka的。群比。总数的以及由此产生的聚合值。

我正试图每天汇总每分钟的事件。

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

我正在使用kafka StreamsgroupBy和aggregate对每天和每家每户的这些事件进行汇总。

通常，由于一天有1440分钟，因此聚合值不应超过1440。此外，不应存在事件量为负值的聚合。

...但它无论如何都会发生，我们不明白代码中有什么错误。

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

java prettyprint-override">
        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）；
• 我们被迫使用Ktable.groupBy（）。聚合（），因为对于已经发布的即时消息，分钟值可以用不同的传感器值重新发布。和每日聚合相应地修改。Stream.groupBy（）。聚合（）没有一个加法器和一个子集。

Answer 1

我认为，实际上有可能计数变成负值。

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

如果当前计数为零，并且在加法之前进行减法运算，则计数将暂时变为负数。