如何将当前事件与kafka流中的旧事件进行比较

Question

我在主题 T1 中有一系列事件，key 作为员工 ID，value 作为员工对象。

例如：

Key: Value (JSON)
     1000: bob,24,department1,etc
     1002: sam,35,department2,etc
     1003: anna,26,department1,etc
     1000: bob,34,department2,etc`

使用 kafka 流，我需要与已处理的旧值进行比较，并确定其是否为 CREATE、UPDATE 并且必须为主题 T2 生成事件说

例如：

Key: Value (JSON)
     1000: CREATE,bob,24,department1,etc
     1002: CREATE,sam,35,department2,etc
     1003: CREATE,anna,26,department1,etc
     1000: UPDATE,bob,34,department2,etc`

如何实现此功能？我可以使用reduce操作吗？如果有，怎么做？

Answer 1

按照 Matthias 的建议，我将使用处理器 API 来实现自定义处理器。

首先，您需要创建一个实现处理器的类，这是您所有自定义逻辑的所在。每次有新记录到达并且您想比较它们时，都会调用 process() 方法。 将 PAPI（处理器 API）逻辑与您的业务逻辑混合并不是您在生产就绪应用程序中所拥有的，但在此示例中，它可以工作。

这看起来像下面这样。

public class EmployeeProcessor implements Processor<Long, JsonNode, Long, JsonNode> {
    private KeyValueStore<Long, JsonNode> kvStore;
    private ProcessorContext<Long, JsonNode> context;


    @Override
    public void init(final ProcessorContext<Long, JsonNode> context) {
        // Initialise the context and state store to use them in the process() method
        this.context = context;
        kvStore = context.getStateStore("EmployeeRecords");
    }

    @Override
    public void process(final Record<Long, JsonNode> record) {
        // Get the stored employee
        JsonNode oldEmployee = kvStore.get(record.key());

        // Transform the Json object to be able to modify it.
        ObjectNode updatedEmployee = (ObjectNode) record.value();

        // If the employee is not present, then forward it as a CREATE
        if (oldEmployee == null) {
            // Example logic to add CREATE property to Json object
            updatedEmployee.put("operation", "CREATE");
            kvStore.put(record.key(), updatedEmployee);
        }
        // If the employee is present, then forward it as a UPDATE
        else {
            // Example logic to add UPDATE property to Json object
            updatedEmployee.put("operation", "UPDATE");
            kvStore.put(record.key(), updatedEmployee);
        }
    }

    @Override
    public void close() {
        // close any resources managed by this processor
        // Note: Do not close any StateStores as these are managed by the library
    }
}

然后您需要添加 State Store 和 Sink 并在创建拓扑时连接它们。应该是这样的：

Topology builder = new Topology();

        // Using a `KeyValueStoreBuilder` to build a `KeyValueStore`.
        StoreBuilder<KeyValueStore<Long, JsonNode>> employeeStateStore =
                Stores.keyValueStoreBuilder(
                        Stores.persistentKeyValueStore("EmployeeRecords"),
                        Serdes.Long(),
                        Serdes.jsonSerde());
        
        // add the source processor node that takes Kafka topic "source-topic" as input
        builder.addSource("Source", "employee-topic")
                // add the WordCountProcessor node which takes the source processor as its upstream processor
                .addProcessor("Process", () -> new EmployeeProcessor(), "Source")
                // add the count store associated with the WordCountProcessor processor
                .addStateStore(employeeStateStore, "Process")
                // add the sink processor node that takes Kafka topic "sink-topic" as output
                // and the WordCountProcessor node as its upstream processor
                .addSink("Sink", "employee-sink-topic", "Process");

请记住，我假设您已经有一个 jsonSerde 类来处理您的 JSON 对象的序列化和反序列化。

查看 Kafka 流文档，其中详细说明了此处理器 API 的工作原理并提供了有用的示例。我在此处提供的示例受到这些文档的启发。

这是最新的 Kafka 流文档的链接，您可以将其用作参考：https://kafka.apache.org/30/documentation/streams/developer-guide/processor-api.html