从kafka阅读时，请使用keyby vs reinterpretaskeyedstream（）

Question

我有一个简单的Flink流处理应用程序（Flink版本1.13）。 Flink应用程序读取了Kakfa，对记录进行了陈述的处理，然后将结果写回Kafka。 从Kafka主题阅读后，我选择使用retinterpretaSkeyedStream（）而不是keyby（）避免进行混乱，因为记录已经在kakfa中进行了分区。用于分区的键是记录的字符串字段（使用默认的kafka分区器）。 Kafka主题有24个分区。

映射类定义如下。它跟踪记录的状态。

public class EnvelopeMapper extends
        KeyedProcessFunction<String, Envelope, Envelope> {
   ...
}

记录的处理如下：

        DataStream<Envelope> messageStream =
                env.addSource(kafkaSource)

        DataStreamUtils.reinterpretAsKeyedStream(messageStream, Envelope::getId)
                .process(new EnvelopeMapper(parameters))
                .addSink(kafkaSink);

平行性为1，代码运行正常。随着平行性大于1（例如4），我正在遇到以下错误：

2022-06-12 21:06:30,720 INFO  org.apache.flink.runtime.executiongraph.ExecutionGraph       [] - Source: Custom Source -> Map -> Flat Map -> KeyedProcess -> Map -> Sink: Unnamed (4/4) (7ca12ec043a45e1436f45d4b20976bd7) switched from RUNNING to FAILED on 100.101.231.222:44685-bd10d5 @ 100.101.231.222 (dataPort=37839).
java.lang.IllegalArgumentException: KeyGroupRange{startKeyGroup=96, endKeyGroup=127} does not contain key group 85

基于堆栈跟踪，当explovemapper class class的验证记录已将记录发送到正确的复制品时，似乎发生了例外。映射器对象。

当使用reinterpretAskeyedStream（）时，记录如何在eventMapper的不同副本中分布？

先感谢您， 艾哈迈德。

更新

在@david Anderson的反馈后 ，替换为reinterpretaSkeyedStream（），用keyby（）。记录的处理如下如下：

        DataStream<Envelope> messageStream =
                env.addSource(kafkaSource)      // Line x
                   .map(statelessMapper1)
                   .flatMap(statelessMapper2);

        messageStream.keyBy(Envelope::getId)
                     .process(new EnvelopeMapper(parameters))
                     .addSink(kafkaSink);

如果keyby（）在从kakfa阅读后立即完成（用“行x”标记）而在状态映射之前立即完成性能（（<）代码> emplowemapper ）。

Answer 1

在

reinterpretAsKeyedStream(
    DataStream<T> stream,
    KeySelector<T, K> keySelector,
    TypeInformation<K> typeInfo)

您的情况下，声称记录已经按照您使用的keyby（keyselector）完全分布。通常情况下，这种记录直接从卡夫卡出来。即使它们是由Kafka中的Key划分的，Kafka分区也不会与Flink的主要组正确相关。

reinterpretaSkeyedStream仅在处理窗口或过程函数的输出之类的情况下直接有用，在此情况下，您知道输出记录是以特定方式分区的。在Kafka上成功使用它可能非常困难：首先，您必须非常谨慎地将数据写入Kafka，或者使用KeySelector做一些棘手的事情，以使其计算的密钥组与键如何与键保持一致。被映射到卡夫卡分区。

一种情况并不困难的一种情况是，是否通过与正在读取数据并使用retinterpretaskeyedstream的下游作业相同的配置来写入Kafka。