java - Kafka Streams 在多个流中拆分 1 个流

Question

我有一个项目来学习使用 Kafka Streams，但我真的很难用它。我正在研究 kafka-streams 版本 1.0.1。我们有一个主主题流，其中包含具有以下样式的消息:

{
    "phenomenonTime" : "2017-04-03T16:08:19.000Z",
    "resultTime" : "2017-04-03T16:08:19.000Z",
    "result" : {
      "Temperature" : 0,
      "Pressure" : 0,
      "Humidity" : 0,
      "Mean altitude" : 0,
      "Mass PM2.5" : 7.4,
      "Mass Error PM2.5" : 1.5,
      "Mass PM10" : 12.3,
      "Mass Error PM10" : 1.5
    }
  }

这是一个 Json 格式，这是我的第一个问题，我不知道 Json Deserializer 或 Serializer 究竟如何使用。

但我的主要目标是创建主题温度、压力、湿度、平均海拔高度......在主要主题的结果字段中，温度主题中具有正确的值。

我如何通过 Kafka Streams 实现这一点？我希望你能帮助我更好地开始使用 Kafka Streams。

编辑:

Hole Message + key(格式化)

    Key c45e9532-9810-11e8-8839-03e1e3365152
    Value { "phenomenonTime" : "2017-04-03T16:08:09.000Z",
 "resultTime" : "2017-04-03T16:08:09.000Z",
 "result" : { "Temperature" : 0,
 "Pressure" : 0, 
"Humidity" : 0, 
"Mean altitude" : 0, 
"Mass PM2.5" : 7.1,
 "Mass Error PM2.5" : 1.5,
 "Mass PM10" : 9.6, "Mass Error PM10" : 1.5 },
 "Datastream@iot.navigationLink" : "http://localhost:8080/FROST-Server/v1.0/Observations('c45e9532-9810-11e8-8839-03e1e3365152')/Datastream", 
"Datastream" : { "unitOfMeasurement" : { "name" : null, "symbol" : null, "definition" : null }, "@iot.id" : "geo.uni-augsburg.de/Fixed-Wing-UAV-1/Datastreams/LOAC_LOCAL_201704031605.mass" }, 
"FeatureOfInterest@iot.navigationLink" : "http://localhost:8080/FROST-Server/v1.0/Observations('c45e9532-9810-11e8-8839-03e1e3365152')/FeatureOfInterest",
 "FeatureOfInterest" : { "@iot.id" : "c458a1a4-9810-11e8-8839-176a6dbe6951" }, "@iot.id" : "c45e9532-9810-11e8-8839-03e1e3365152", "@iot.selfLink" : "http://localhost:8080/FROST-Server/v1.0/Observations('c45e9532-9810-11e8-8839-03e1e3365152')" }

未格式化:

Key c45e9532-9810-11e8-8839-03e1e3365152
Value { "phenomenonTime" : "2017-04-03T16:08:09.000Z", "resultTime" : "2017-04-03T16:08:09.000Z", "result" : { "Temperature" : 0, "Pressure" : 0, "Humidity" : 0, "Mean altitude" : 0, "Mass PM2.5" : 7.1, "Mass Error PM2.5" : 1.5, "Mass PM10" : 9.6, "Mass Error PM10" : 1.5 }, "Datastream@iot.navigationLink" : "http://localhost:8080/FROST-Server/v1.0/Observations('c45e9532-9810-11e8-8839-03e1e3365152')/Datastream", "Datastream" : { "unitOfMeasurement" : { "name" : null, "symbol" : null, "definition" : null }, "@iot.id" : "geo.uni-augsburg.de/Fixed-Wing-UAV-1/Datastreams/LOAC_LOCAL_201704031605.mass" }, "FeatureOfInterest@iot.navigationLink" : "http://localhost:8080/FROST-Server/v1.0/Observations('c45e9532-9810-11e8-8839-03e1e3365152')/FeatureOfInterest", "FeatureOfInterest" : { "@iot.id" : "c458a1a4-9810-11e8-8839-176a6dbe6951" }, "@iot.id" : "c45e9532-9810-11e8-8839-03e1e3365152", "@iot.selfLink" : "http://localhost:8080/FROST-Server/v1.0/Observations('c45e9532-9810-11e8-8839-03e1e3365152')" }

但是 Datastream@iot.navigationLink、Datastream .... 不是重要的。但Key必须相同。

这就是它的样子 (https://i.imgur.com/zvwf3g7.png)

孔导出流:

https://pastebin.com/PUfhL8fK

Kafka 客户端示例:

https://pastebin.com/y4k7fQgz

Answer 1

为此，您需要为每个所需的目标主题创建多个 KStream 对象。要从主 json 中提取所需字段，请在 kStream 上使用 mapValues 方法。为了简化使用 json 值的工作，您可以使用 spring-kafka 库中的 JsonSerde(groupId:org.springframework.kafka，artifactId:spring-kafka)。

温度和压力主题的示例(并对每个所需的目标主题执行相同的操作):

Map<String, String> streamProperties = new HashMap<>();
streamProperties.put("bootstrap.servers", "localhost:9092");
streamProperties.put("key.serde", "org.apache.kafka.common.serialization.Serdes$StringSerde");
streamProperties.put("value.serde", "org.apache.kafka.common.serialization.Serdes$StringSerde");

Map<String, String> streamProperties1 = new HashMap<>(streamProperties);
streamProperties1.put("application.id", "temperature");
Map<String, String> streamProperties2 = new HashMap<>(streamProperties);
streamProperties2.put("application.id", "pressure");

Class<Map<String, Object>> genericMapClass = (Class) Map.class;
Consumed<String, Map<String, Object>> consumed = Consumed.with(Serdes.String(), new JsonSerde<>(genericMapClass));
Produced<String, Map<String, Object>> produced = Produced.with(Serdes.String(), new JsonSerde<>(genericMapClass));

StreamsBuilder streamBuilder1 = new StreamsBuilder();
KStream<String, Map<String, Object>> temperatureKStream = streamBuilder1.stream("mainSourceTopic", consumed);
temperatureKStream.mapValues((generalDetails) -> {
    Object temperatureValue = ((Map) generalDetails.get("result")).get("Temperature");
    Map<String, Object> temperatureMessageDetails = new HashMap<>();
    temperatureMessageDetails.put("Temperature", temperatureValue);
    temperatureMessageDetails.put("phenomenonTime", generalDetails.get("phenomenonTime"));
    temperatureMessageDetails.put("resultTime", generalDetails.get("resultTime"));
    System.out.println("temperatureMessageDetails: " + temperatureMessageDetails);
    return temperatureMessageDetails;
}).to("temperatureTopic", produced);

StreamsBuilder streamBuilder2 = new StreamsBuilder();
KStream<String, Map<String, Object>> pressureKStream = streamBuilder2.stream("mainSourceTopic", consumed);
pressureKStream.mapValues((generalDetails) -> {
    Object pressureValue = ((Map) generalDetails.get("result")).get("Pressure");
    Map<String, Object> pressureMessageDetails = new HashMap<>();
    pressureMessageDetails.put("Pressure", pressureValue);
    pressureMessageDetails.put("phenomenonTime", generalDetails.get("phenomenonTime"));
    pressureMessageDetails.put("resultTime", generalDetails.get("resultTime"));
    System.out.println("pressureMessageDetails: " + pressureMessageDetails);
    return pressureMessageDetails;
}).to("pressureTopic", produced);

StreamsConfig streamsConfig1 = new StreamsConfig(streamProperties1);
KafkaStreams kafkaStreams1 = new KafkaStreams(streamBuilder1.build(), streamsConfig1);
kafkaStreams1.start();

StreamsConfig streamsConfig2 = new StreamsConfig(streamProperties2);
KafkaStreams kafkaStreams2 = new KafkaStreams(streamBuilder2.build(), streamsConfig2);
kafkaStreams2.start();

Runtime.getRuntime().addShutdownHook(new Thread(() -> {
    kafkaStreams1.close();
    kafkaStreams2.close();
}));