Kafka流|连接流的聚合

Question

我想连接两个主题流（左连接），并在连接的流上进行基于窗口的聚合。然而，聚合将某些消息计数两倍，因为在连接期间，根据正确主题中的延迟，某些消息将发出两倍。以下是POC的代码。

        StreamsBuilder builder = new StreamsBuilder();
        KStream<Long, BidMessage> bidStream = builder.stream("bid", Consumed.with(new LongSerde(), new BidMessageSerde()).withTimestampExtractor(new BidMessageTimestampExtractor()));
        KStream<Long, ClickMessage> clickStream = builder.stream("click", Consumed.with(new LongSerde(), new ClickMessageSerde()).withTimestampExtractor(new ClickMessageTimestampExtractor()));

        KStream<String, BidMessage> newBidStream = bidStream.selectKey((key, value) -> value.getRequestId());
        KStream<String, CLickMessage> newClickStream = impStream.selectKey((key, value) -> value.getRequestId());

        KStream<String, BidMergedMessage> result = newBidStream.leftJoin(newImpStream,
                getValueJoiner(),
                JoinWindows.of(Duration.ofSeconds(30)).grace(Duration.ofSeconds(0)),
                Joined.with(Serdes.String(), new BidMessageSerde(), new ClickMessageSerde()));

        result.groupBy((key, value) -> "" + value.getClientId(), Grouped.with(Serdes.String(), newBidMergedSerde()))
                .windowedBy(TimeWindows.of(Duration.ofSeconds(30)).grace(Duration.ofSeconds(40)))
                .aggregate(() -> new AggResult(0, 0), (key, value, aggregate) -> {
                    if (value.getClickId() != null) {
                        aggregate.clicks_++;
                    }
                    aggregate.bids_++;
                    return aggregate;
                }, Materialized.with(Serdes.String(),new AggResultJsonSerde()))
                .suppress(Suppressed.untilWindowCloses(Suppressed.BufferConfig.unbounded()))
                .toStream()
                .foreach((key, value) -> {
                    logger.info("{}-{}, clientId : {}, Value: {}", new Date(key.window().start()), new Date(key.window().end()),key.key(), value);
                });

        final KafkaStreams streams = new KafkaStreams(builder.build(), props);
        streams.start();

它是否可以修复以避免因连接而重复？

Answer 1

我可以想出两个选择。

>

首先要尝试的是在KStream中添加一个filter

使用内部联接newBidStream。加入（newImpStream…

-比尔

Answer 2

我不确定我是否完全遵循了你的要求，但问题似乎是你只想在没有与RHS合并或加入成功的情况下计算出价。但是由于RHS主题中的一些缓慢，您偶尔会得到两个结果，第一个是“未合并”，然后在RHS记录到达时进行合并。

您可以在结果KStream上添加TransformValues运算符，并使用状态存储来跟踪传入的记录。当一个成功的连接中出现了一个副本时，可以查看statestore并删除带有空RHS的记录（如果存在），然后转发正确的连接结果。

为了转发没有成功连接的记录，您可以考虑使用<代码>标点>（）/代码>定期地通过该存储并发出没有匹配的记录，并在您感觉到连接发生的时间内已经在StestStor中。

Kafka教程中的本教程也可以作为指南。