基于公共字段的Kstream连接

Question

我们希望基于公共字段（主键）执行Kstream Kstream连接。目前，使用下面的代码，我们得到的结果是只合并了两个流，没有任何主键约束。

val userRegions: KStream[String, String] = builder.stream(inputTopic1)
val regionMetrics: KStream[String, String] = builder.stream(inputTopic2)


userRegions.join(regionMetrics)(
  ((regionValue, metricValue) => regionValue + "/" + metricValue),
  JoinWindows.of(Duration.ofMinutes(5L))
).to(outputTopicName)

您能建议如何根据公共字段/列连接2个流吗。

Answer 1

为了加入基于公共字段/列的2个流，您可以使用selectKey（）函数，我将为您提供一个片段，可以帮助您。

val userRegions: KStream[String, String] = builder.stream(inputTopic1)
val regionMetrics: KStream[String, String] = builder.stream(inputTopic2)

// New code
val userRegionsWithKeys = userRegions.selectKey(new ValueMapper (String key, String Value) {
   // create your key here and return it
   // Please the syntax for Scala
    @override
    void apply (String key, String value) {
       return "key that you want" 
    }
});

val regionMetricsWithKeys = regionMetrics.selectKey(new ValueMapper (String key, String Value) {
   // create your key here and return it
   // Please the syntax for Scala
    @override
    void apply (String key, String value) {
       return "key that you want" 
    }
});


userRegionsWithKeys .join(regionMetricsWithKeys )(
  ((regionValue, metricValue) => regionValue + "/" + metricValue),
  JoinWindows.of(Duration.ofMinutes(5L))
).to(outputTopicName)

我希望这个解决方案能帮助你。

非常感谢。