Kafka-Streams加入2个主题与JSON值背压机制？

Question

我正在学习Kafka流，并尝试实现以下几点：

创建了2个Kafka主题（比如topic1、topic2)，键为null，值为JSONString。来自topic1的数据（无重复项）在Topic2中有多个匹配项。即。topic1具有一些主流数据，当与Topic2连接时可以生成新的多数据流。

示例：

topic1={"name": "abc", "age":2}, {"name": "xyz", "age":3} and so on.
topic2={"name": "abc", "address"="xxxxxx"}, {"name": "abc", "address"="yyyyyy"}, {"name": "xyz", "address"="jjjjjj"}, {"name": "xyz", "address"="xxxkkkkk"}

我尝试了一段带有5分钟窗口的KStream代码，但看起来我无法在流中保存topic1数据。

请帮我做出正确的选择并加入。我正在使用来自Confluent和Docker实例的Kafka。

public void run() {
        final StreamsBuilder builder = new StreamsBuilder();
        final Serde<JsonNode> jsonSerde = Serdes.serdeFrom(new JsonSerializer(), new JsonDeserializer());
        final Consumed<String, JsonNode> consumed = Consumed.with(Serdes.String(), jsonSerde);

        // Hold data from this topic to 30 days
        KStream<String, JsonNode> cs = builder.stream("topic1", consumed);
        cs.foreach((k,v) -> {
            System.out.println( k + " --->" + v);
        });

        // Data is involved in one time process.
        KStream<String, JsonNode> css = builder.stream("topic2", consumed);
        css.foreach((k,v) -> {
            System.out.println( k + " --->" + v);
        });

        KStream<String, JsonNode> resultStream = cs.leftJoin(css,
                valueJoiner,
                JoinWindows.of(TimeUnit.MINUTES.toMillis(5)),
                Joined.with(
                        Serdes.String(), /* key */
                        jsonSerde,       /* left value */
                        jsonSerde)       /* right value */
        );

        resultStream.foreach((k, v) -> {
            System.out.println("JOIN STREAM: KEY="+k+ ", VALUE=" + v);
        });

        KafkaStreams streams = new KafkaStreams(builder.build(), properties);
        streams.start();
    }

Answer 1

Kafka中的连接总是基于键(*)。因此，要使任何连接工作，您需要在执行实际连接之前将要连接的字段提取到键中（唯一的部分例外是KStream-GlobalKTable连接）。在您的代码示例中，您不会得到任何结果，因为所有记录都有一个null键，因此无法联接。

对于连接本身，对于您的用例来说，KStream-KTable连接似乎是正确的选择。要使此操作成功，您需要：

1. 为topic1正确设置联接键，并将数据写入一个附加主题（我们称之为topic1keyed)
2. 将topic1keyed作为表读取
3. 为topic2
正确设置联接键
4. 将topic2与Ktable
连接

有关join语义的全部详细信息，请参阅这篇博文：https://www.confluent.io/blog/crossing-streams-joins-apache-kafka/

(*)更新：