KafkaStream.KTable如何以（压缩）KV风格将数据写入kafka主题答案

【问题标题】：How does KafkaStream.KTable write data to kafka topic in (compacted) KV styleKafkaStream.KTable如何以（压缩）KV风格将数据写入kafka主题
【发布时间】：2018-03-02 17:21:30
【问题描述】：

在 Kafka(0.11.0.1) Streams 中，一个演示应用 Play with a Streams Application

// Serializers/deserializers (serde) for String and Long types
final Serde<String> stringSerde = Serdes.String();
final Serde<Long> longSerde = Serdes.Long();

// Construct a `KStream` from the input topic "streams-plaintext-input", where message values
// represent lines of text (for the sake of this example, we ignore whatever may be stored
// in the message keys).
KStream<String, String> textLines = builder.stream(stringSerde, stringSerde, "streams-plaintext-input");

KTable<String, Long> wordCounts = textLines
    // Split each text line, by whitespace, into words.
    .flatMapValues(value -> Arrays.asList(value.toLowerCase().split("\\W+")))

    // Group the text words as message keys
    .groupBy((key, value) -> value)

    // Count the occurrences of each word (message key).
    .count("Counts")

    // Store the running counts as a changelog stream to the output topic.
    wordCounts.to(stringSerde, longSerde, "streams-wordcount-output");

第 5 步，处理完一些数据后，我们可以在 sink topic streams-wordcount-output 中看到压缩的 KV 对（例如 streams 2），

> bin/kafka-console-consumer.sh --bootstrap-server localhost:9092
    --topic streams-wordcount-output \
    --from-beginning \
    --formatter kafka.tools.DefaultMessageFormatter \
    --property print.key=true \
    --property print.value=true \
    --property key.deserializer=org.apache.kafka.common.serialization.StringDeserializer \
    --property value.deserializer=org.apache.kafka.common.serialization.LongDeserializer 

all     1 
streams 1 
lead    1 
to      1 
kafka   1 
hello   1 
kafka   2 
streams 2

问题是上述数据中的KTable wordCounts如何以Key-Value的方式将数据写入主题streams-wordcount-output？

主题 streams-wordcount-output 的选项 cleanup.policy 似乎是默认值，delete，而不是 compact(via bin/kafka- configs.sh)

【问题讨论】：

标签： java apache-kafka apache-kafka-streams

【解决方案1】：

所有输入和输出主题都“超出”Kafka Streams 的范围。创建和配置这些主题是用户的责任。

因此，您的主题"streams-wordcount-output" 将具有您在创建主题时指定的配置。

参见。 https://docs.confluent.io/current/streams/developer-guide.html#managing-topics-of-a-kafka-streams-application

【讨论】：

指南中，第三步创建了streams-wordcount-output，其中cleanup.policy是默认delete