如何使用KTable作为参考数据来更新KStream？

Question

我有一个Kafka主题，包含Json格式的数据：

{"id": "A", "country": "France"}
{"id": "B", "currency": "£"}

我想用类似“参考表”的东西来规范内容：

country ( "France" ) -> "FR"
currency ( "£" ) -> "GBP"

为了输出：

{"id": "A", "country": "FR"}
{"id": "B", "currency": "GBP"}

我认为这是使用KTable存储参考数据的典型用例。但我在实施上有点纠结。

当前状态

摄取参考数据

在Kafka上创建的专用主题：poc映射在

主题提供了示例Json数据：

{"mapping":"ccy",     "from":"£",      "to":"GBP"}
{"mapping":"country", "from":"France", "to":"FR"}

在对键和值进行返工后，在KTable中摄入的数据：

         KStream<String, String> mappingStream = builder
                .stream("poc-mapping-in",consumed)
                .map(
                     (key, value) -> KeyValue.pair(
                         value.get("mapping")+"#"+value.get("from"), 
                         value.get("to").asText())
         );

         KGroupedStream<String, String> mappingGroupedStream = mappingStream.groupByKey(
                 Serialized.with(Serdes.String(),Serdes.String() ));


         KTable<String,String> mappingTable = mappingGroupedStream.aggregate(
                () -> "", //initializer 
                (aggKey, newValue, aggValue) -> newValue, // adder 
                Materialized.<String, String, KeyValueStore<Bytes, byte[]>>as("ReferenceStore")
                    .withValueSerde(Serdes.String())
                    .withKeySerde(Serdes.String())
                );

           // Testing
           mappingTable.toStream().to("poc-mapping-in-content", 
                Produced.with(Serdes.String(), Serdes.String()));

在主题poc-maping-in-content中，我得到以下行：

"currency"#"£"      GBP
"country"#"France"  FR

这看起来像我所期望的。双引号很奇怪，但这并不妨碍我走得更远。

数据被/应该存储在名为ReferenceStore的本地存储中。

获取业务流

在Kafka上创建的主题：poc原始事件

主题提供了示例Json数据：

{"id": "A", "country": "France"}
{"id": "B", "currency": "£"}

在KStream中摄取的数据：

  final Consumed<String, JsonNode> consumed = Consumed.with(Serdes.String(), jsonSerde);
  KStream<String, JsonNode> businessData = builder.stream("poc-raw-events", consumed);

从这里我不知道该怎么办。从技术上讲，我知道如何更新JsonNode中的属性。因此，我尝试使用foreach在KStream上循环，方法如下：

    businessData.foreach(new ForeachAction<String, JsonNode>()  {
        public void apply(String k, JsonNode v) {
            System.out.println(k+ " : " +v);
                    if (v==null) {System.out.println("NULL detected"); return;}
            Iterator<Entry<String, JsonNode>> fields = v.fields();
            int i=0;
            while (fields.hasNext()) {
                i++;
                Entry<String, JsonNode> next = fields.next();
                System.out.println(k+ " field #"+i+" : " +next.getKey() + " -- " + next.getValue());

                String key = next.getKey() + "#" + next.getValue());
//              ((ObjectNode) v).put(next.getKey(), "  WHAT HERE ??? ");

            }

        }
    });

我的想法是替换“这里是什么？” 位于最后一行中，数据显示在参考KTable中。但是怎么做？？？

• 我没有在KTable上找到类似. findByKey（）的东西。
• 我不知道如何访问参考存储本地存储，因为访问它的方式类似于myKafkaStream.store（...），此时myKafkaStream还没有开始，甚至还没有构建。

我考虑的另一种方法是使用KStream leftJoin KTable功能。但我在某个地方读到过（我没有书签…）要做到这一点，我们应该在两个kTable中使用相同的密钥。但在我的例子中，在Json方面，我不处理join键，而是处理一个简单的属性。

你将如何实现这一点？

Answer 1

如果ReferenceTable有一个与数据匹配的键。getAltKey（）

streamToMap.selectKey((originalKey, data) -> data.getAltKey()).leftJoin(referenceKTable, valueJoiner)

我们可以做到这一点。valueJoiner（或lambda）的实现必须结合这两个输入。

Answer 2

由于使用引用数据，我认为您要考虑使用的是<代码> GualCalabe< /Cord>。每个KafkaStreams实例都会完全复制GlobalKTable，并显式创建该表，以保存上述用例的参考数据。

KStream GlobalKTable联接的独特之处在于，您可以使用流的KeyValue映射到GlobalKTable的键。因此，只要您可以从JsonNode中提取属性，就应该能够与GlobalKTable中的相应记录联接