Apache Flink动态接收器数

Question

我正在使用Apache Flink和Kafka消费者阅读Kafka主题中的一些值。我还有一个通过读取文件获得的流。

根据收到的值，我想就不同的Kafka主题编写这个流。

基本上，我有一个与许多孩子联系在一起的领导者的网络。对于每个孩子，领导者需要编写在孩子特定的Kafka主题中读取的流，以便孩子可以阅读它。当子级启动时，它会在从领导者读取的Kafka主题中注册自己。问题是我不知道我有多少个孩子。

例如，我从Kafka主题中阅读1，我只想在一个名为Topic1的Kafka主题中编写流。我读了1-2，我想写两个Kafka主题（Topic1和Topic2）。

我不知道这是否可能，因为为了写这个主题，我使用了Kafka生产者和addSink方法，据我理解（从我的尝试来看），Flink似乎需要知道汇的数量先验。

但是，难道没有办法获得这种行为吗？

Answer 1

现在不建议使用KeyedSeriazationSchema。相反，您必须使用"KafkaSeriazationSchema"

同样可以通过重写serialize方法来实现。

    public ProducerRecord<byte[], byte[]> serialize(
String inputString, @Nullable Long aLong){ 
        return new ProducerRecord<>(customTopicName,
 key.getBytes(StandardCharsets.UTF_8), inputString.getBytes(StandardCharsets.UTF_8));
}

Answer 2

如果我能很好地理解你的问题，我认为你可以用一个接收器来解决它，因为你可以根据正在处理的记录来选择Kafka主题。似乎源代码中的一个元素可能会写入多个主题，在这种情况下，您需要一个FlatMapFunction将每个源代码记录复制N次（每个输出主题一个）。我建议将（topic，record）作为一对（akaTuple2）输出。

DataStream<Tuple2<String, MyValue>> stream = input.flatMap(new FlatMapFunction<>() {
    public void flatMap(MyValue value, Collector<Tupple2<String, MyValue>> out) {
        for (String topic : topics) {
            out.collect(Tuple2.of(topic, value));
        }
    }
});

然后，您可以使用前面通过创建FlinkKafkaProducer计算出的主题，该主题带有一个KeyedSerializationSchema，在其中实现getTargetTopic，以返回对中的第一个元素。

stream.addSink(new FlinkKafkaProducer10<>(
        "default-topic",
        new KeyedSerializationSchema<>() {
            public String getTargetTopic(Tuple2<String, MyValue> element) {
                return element.f0;
            }
            ...
        },
        kafkaProperties)
);