加入两个 Kafka 流

Question

尝试加入两个 Kstream，但出现类型不匹配错误，下面是代码 sn-p。

  KStream<String, String> longCounts = netExpence.join(netIncome, (key1, key2) -> key1 + "/" + key2,                    JoinWindows.of(joinWindowSizeMs).until(windowRetentionTimeMs),stringSerde, stringSerde, stringSerde);

出现的错误是Type mismatch: cannot convert from String to R 这是加入 kstream 的语法join(KStream<K,VO> otherStream, ValueJoiner<? super V,? super VO,? extends VR> joiner, JoinWindows windows, Joined<K,V,VO> joined)

请解释一下ValueJoiner<? super V,? super VO,? extends VR> 的作用。谢谢

Answer 1

使用匹配记录的两个值调用ValueJoiner，并发出一个连接结果值。

// key type must be the same for a join
// value type can be different
KStream<KeyType, ValueType1> stream1 = ...
KStream<KeyType, ValueType2> stream2 = ...

KStream<KeyType, OutputType> joined = stream1.join(stream2, ...);

因此，ValueJoiner 必须将 ValueType1 作为第一个泛型 (? super V)，并将 ValueType2 作为第二个泛型 (? super VO)。对于第三个泛型 (? extend VR)，您指定输出类型（即上面示例中的 OutputType）。

更新

您还需要为运行时配置正确的 Serdes。如果所有类型都相同，最好通过 StreamsConfig 为键和/或值相应地设置默认 serdes。否则，您可以覆盖每个运算符的默认 Serdes：

• https://docs.confluent.io/current/streams/developer-guide/config-streams.html#default-key-serde
• https://docs.confluent.io/current/streams/developer-guide/datatypes.html