Kafka Streams Store 值的最新组合表

Question

我正在尝试将最新出现的值组合存储在表中。 因此，正在向患者和传感器发送消息：

{ patient: "A", sensor: "X1", measurementTime: 1625501773},
{ patient: "A", sensor: "Y1", measurementTime: 1625501773},
{ patient: "B", sensor: "X2", measurementTime: 1625501772}

我想更新每个患者的列表，每个传感器的最新测量时间。有人知道如何使用 Kafka Streams API 做到这一点吗？

我在想类似以下的事情：

KeyValueBytesStoreSupplier PatientSensorSupplier = Stores.inMemoryKeyValueStore("latest-patient-sensor-occurrence");

KStream<String, SensorData> messagesStream = builder.stream(
   Topic.Message.getKafkaName(),
   Consumed.with(Serdes.String(), new NoSqlSerde())
        .withTimestampExtractor((k, partitionTime) -> {
            Long measurementTime = ((SensorDataPojo) k.value()).getMeasurementTime();
            if (measurementTime != null) {
                return measurementTime;
            }
            return partitionTime;
        })
);

KStream<String, String> patientsWithSensors = messagesStream.map(
    (key, sensorDataPojo) ->
        new KeyValue<>(
            noSqlSensorDataPojo.getPatientId(),
            noSqlSensorDataPojo.getSensorId()
        )
);

KTable<String, SensorListSerde> patientsWithSensorsTable = patientsWithSensors.toTable(
   Named.as("patients-sensors-occurrences-table"),
   Materialized.<String, String>as(storeSupplier)
       .withKeySerde(Serdes.String())
       .withValueSerde(sensorListSerde)
);

但是使用它我无法弄清楚如何在表格中正确存储每个患者每个传感器的最新记录的时间戳。我假设必须有更好的方法来解决这个问题。

Answer 1

您现在的操作方式KTable 将PatientId 作为键，SensorId 作为值。这意味着它将始终包含每位患者最后一次测量的SensorId。

您现在已经知道如何从数据中提取时间戳，您可以定义一个包含timestamp 和sensorId 的pojo，并将其用作值。为此，您必须使用 JsonPOJOSerializer。

但是，我不确定为什么您首先希望在 KTable 中最近出现您的值。你想用这个做什么？

• 如果您想进行某种聚合，也可以在流中进行。
• 如果您想将最新数据持久保存在数据库中，最好使用 Kafka Connect。

文档是here 和here

---

编辑：

我刚刚偶然发现了 Kafka Streams 中的 Stores.timestampedKeyValueStoreBuilder() 方法。如果您将第一行替换为

TimeStampedKeyValueBytesStore PatientSensorSupplier = 
Stores.timestampedKeyValueStoreBuilder(Stores.inMemoryKeyValueStore("latest-patient-sensor-occurrence"), Serdes.String(), sensorListSerde, Serdes.).build();

它可能只是工作。