Kafka比较键的连续值

Question

我们正在构建一个应用程序来从传感器获取数据。数据流传输到Kafka，消费者将从Kafka发布到不同的数据商店。每个数据点都有多个表示传感器状态的属性。

在其中一个消费者中，我们希望仅当值发生更改时才将数据发布到数据存储。例如，如果有温度传感器，每10秒轮询一次数据，我们希望收到如下数据：

----------------------------------------------------------------------
Key                Value
----------------------------------------------------------------------
Sensor1            {timestamp: "10-10-2019 10:20:30", temperature: 10}
Sensor1            {timestamp: "10-10-2019 10:20:40", temperature: 10}
Sensor1            {timestamp: "10-10-2019 10:20:50", temperature: 11}

在上述情况下，只应发布第一条记录和第三条记录。

为此，我们需要某种方法来比较键的当前值与具有相同键的先前值。我认为这在KTable或KStream中应该是可能的，但找不到示例。

任何帮助都会很棒！

Answer 1

如果要对Kafka流执行此操作，必须使用处理器API。

您需要使用状态存储实现自定义变压器。对于每条消息，您应该在状态存储中搜索值，如果它已更改或不存在，您应该返回新值，否则为null。除此之外，您还应该将该值保存在状态存储中（KeyValueStore：：put（…） ）

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

Answer 2

您可以使用Kafka流处理器API。可以将本地键值存储设置为状态上下文。为获取的每个记录调用process函数。

在process function中，您可以对照存储的最后一个值进行检查，并根据业务逻辑（在您的情况下，比较温度值）接受或拒绝最新记录。

在标点功能中，您可以按时间表将记录转发给消费者。请参阅下面的示例代码（不带标点符号）

java prettyprint-override">public class SensorProcessor implements Processor<String, String> {

    private ProcessorContext context;
    private KeyValueStore<String, String> kvStore;

    @Override
    @SuppressWarnings("unchecked")
    public void init(ProcessorContext context) {
        // keep the processor context locally because we need it in punctuate() and commit()
        this.context = context;

        // retrieve the key-value store named "SensorData"
        kvStore = (KeyValueStore) context.getStateStore("SensorData");

        // schedule a punctuate() method every second based on event-time
      
    }

    @Override
    public void process(String sensorName, String sensorData) {
      
        String oldValue = this.kvStore.get(sensorName);

        if (oldValue == null) {
            this.kvStore.put(sensorName, sensorData);
        } else {
            //Put the business logic for comparison
            //compare temperatures
            //if required put the value
            this.kvStore.put(sensorName, sensorData);

            //Forward it o consumer
            context.forward(sensorName, sensorData);
        }
        context.commit();
    }

    @Override
    public void close() {
        // nothing to do
    }
}

Answer 3

下面是一个如何使用KStream#transformValues（）解决此问题的示例。

java prettyprint-override">StreamsBuilder builder = new StreamsBuilder();
StoreBuilder<KeyValueStore<String, YourValueType>> keyValueStoreBuilder =
    Stores.keyValueStoreBuilder(Stores.persistentKeyValueStore(stateStoreName),
                                Serdes.String(),
                                YourValueTypeSerde());
builder.addStateStore(keyValueStoreBuilder);
stream = builder.stream(INPUT_TOPIC, Consumed.with(Serdes.Integer(), YourValueTypeSerde()))
    .transformValues(() -> new ValueTransformerWithKey<String, YourValueType, YourValueType>() {
        private KeyValueStore<String, YourValueType> state;

        @Override
        public void init(final ProcessorContext context) {
            state = (KeyValueStore<String, YourValueType>) context.getStateStore(stateStoreName);}

        @Override
        public YourValueType transform(final String key, final YourValueType value) {
            YourValueType prevValue = state.get(key);
            if (prevValue != null) {
                if (prevValue.temperature() != value.temperature()) {
                    return prevValue;
                }
            } else {
                state.put(key, value);
            }
            return null;
       }

       @Override
       public void close() {}
    }, stateStorName))
    .to(OUTPUT_TOPIC);

将该记录与状态存储中存储的上一个记录进行比较。如果温度不同，则从状态存储返回记录，并将当前记录存储在状态存储中。如果温度相等，则丢弃当前记录。