从流中调用 Kafka 交互式查询

Question

我对从 Stream 内部调用交互式查询有特殊要求。这是因为我需要创建一个新的 Stream，它应该在 State Store 中包含数据。以下截断代码：

tempModifiedDataStream.to(topic.getTransformedTopic(), Produced.with(Serdes.String(), Serdes.String()));

GlobalKTable<String, String> myMetricsTable = builder.globalTable(
    topic.getTransformedTopic(),
    Materialized.<String, String, KeyValueStore<Bytes, byte[]>>as(
            topic.getTransformedStoreName() /* table/store name */)
        .withKeySerde(Serdes.String()) /* key serde */
        .withValueSerde(Serdes.String()) /* value serde */
);

KafkaStreams streams = new KafkaStreams(builder.build(), kStreamsConfigs());

KStream<String, String> tempAggrDataStream = tempModifiedDataStream
    .flatMap((key, value) -> {
        try {
            List<KeyValue<String, String>> result = new ArrayList<>();

            ReadOnlyKeyValueStore<String, String> keyValueStore =
                streams .store(
                    topic.getTransformedStoreName(),
                    QueryableStoreTypes.keyValueStore());

在最后一行中，要访问 State Store，我需要拥有 KafkaStreams 对象，并且当我创建 KafkaStreams 对象时拓扑已完成。这种方法的问题在于“tempAggrDataStream”因此不是拓扑的一部分，并且该部分代码不会被执行。而且我不能移动下面的 KafkaStreams 定义，否则我不能调用交互式查询。

我对 Kafka Streams 有点陌生；这对我来说是不是很傻？

Answer 1

如果你想实现每次数据修改后主题内容的全部发送，我认为你应该使用Processor API。

您可以使用状态存储创建org.apache.kafka.streams.kstream.Transformer。 对于每个处理消息，它将更新状态存储并将所有内容发送到下游。 这不是很有效，因为它将为每个处理消息转发主题/状态存储的全部内容（可能是数千、数百万条记录）。

如果您只需要 latest 值，将主题 cleanup.policy 设置为 compact 就足够了。而从其他站点使用KTable，它给出了表的抽象（流的快照）

用于转发状态存储的全部内容的示例 Transformer 代码如下。整个工作在transform(String key, String value)方法中完成。

public class SampleTransformer
        implements Transformer<String, String, KeyValue<String, String>> {

    private String stateStoreName;
    private KeyValueStore<String, String> stateStore;
    private ProcessorContext context;

    public SampleTransformer(String stateStoreName) {
        this.stateStoreName = stateStoreName;
    }

    @Override
    @SuppressWarnings("unchecked")
    public void init(ProcessorContext context) {
        this.context = context;
        stateStore = (KeyValueStore) context.getStateStore(stateStoreName);
    }

    @Override
    public KeyValue<String, String> transform(String key, String value) {
        stateStore.put(key, value);
        stateStore.all().forEachRemaining(keyValue -> context.forward(keyValue.key, keyValue.value));
        return null;
    }

    @Override
    public void close() {

    }
}

有关处理器 API 的更多信息可以找到：

• https://docs.confluent.io/current/streams/developer-guide/processor-api.html
• https://kafka.apache.org/documentation/streams/developer-guide/processor-api.html

如何将处理器 API 与 Stream DSL 结合可以找到：

• https://kafka.apache.org/documentation/streams/developer-guide/dsl-api.html#applying-processors-and-transformers-processor-api-integration