Kafka流中的状态过滤/平坦MapValue？

Question

我正在尝试编写一个简单的Kafka Streams应用程序（目标是Kafka 2.2/Confluent 5.2），将一个至少有一次语义的输入主题转换为一个恰好只有一次的输出流。我想对以下逻辑进行编码：

• 对于具有给定密钥的每条消息：

（这是基于我们从上游系统获得的订购保证来保证提供正确的结果；我不想在这里做任何神奇的事情。）

起初，我以为我可以用Kafka StreamsflatMapValues操作符来实现这一点，它可以让你用同一个键将每个输入消息映射到零个或多个输出消息。然而，该文件明确警告：

这是一个无状态的逐条记录操作（参见有状态值转换的变换值（ValueTransformerProviier，String...））。

这听起来很有希望，但是transformValues留档并没有明确说明如何在每个输入消息中发出零个或一个输出消息。除非这就是示例中的//或null的意思？

平变换看起来也很有希望，但是我不需要操作密钥，如果可能的话，我希望避免重新分区。

有人知道如何正确html" target="_blank">执行这种过滤吗？

Answer 1

如上所述，您可以使用变压器来实现有状态操作。为了不向下游传播消息，您需要从变换方法返回空值，这在变压器java文档中提到。您可以通过进程Context.forward（键，值）来管理传播。下面提供了简化的示例

kStream。变换（（）-

public class DemoTransformer implements Transformer<String, String, KeyValue<String, String>> {
    private ProcessorContext processorContext;
    private String stateStoreName;
    private KeyValueStore<String, String> keyValueStore;

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

    @Override
    public void init(ProcessorContext processorContext) {
        this.processorContext = processorContext;
        this.keyValueStore = (KeyValueStore) processorContext.getStateStore(stateStoreName);
    }

    @Override
    public KeyValue<String, String> transform(String key, String value) {
        String existingValue = keyValueStore.get(key);
        if (/* your condition */) {
            processorContext.forward(key, value);
            keyValueStore.put(key, value);
        }

        return null;
    }

    @Override
    public void close() {
    }
}