使用Flink动态创建和更改Kafka主题

Question

我用Flink来读写来自不同Kafka主题的数据。具体来说，我使用的是FlinkKafkaConsumer和FlinkkafKapProducer。

我想知道是否有可能根据我程序中的逻辑或记录本身的内容，将我正在阅读和写作的Kafka主题更改为“即时”。

例如，如果读取带有新字段的记录，我希望创建一个新主题，并开始将带有该字段的记录转移到新主题。

谢谢。

Answer 1

更多地考虑需求，

第一步是——你将从一个主题开始（为了简单起见），在运行时根据提供的数据生成更多主题，并将相应的消息定向到这些主题。这是完全可能的，不会是一个复杂的代码。使用ZkClient API检查主题名称是否存在，如果不存在，则使用新名称创建一个模型主题，并开始通过与此新主题相关的新生产者将消息推送到模型主题中。无需重新启动作业即可生成指向特定主题的消息。

你最初的消费者成为制作人（新主题）消费者（旧主题）

第二步是-您要使用新主题的消息。一种方法可能是完全创造一份新工作。您可以通过最初创建线程池并向它们提供参数来实现这一点。

同样要更加小心，如果出现循环错误，更多的自动化可能会导致集群过载。考虑一下，如果输入数据不受控制或只是不干净，一段时间后可能会创建太多主题。正如上面评论中提到的，可能会有更好的体系结构方法。

Answer 2

如果您的主题遵循通用命名模式，例如"topic-n*"，您的Flink Kafka消费者可以在添加到Kafka时自动读取"topic-n1"、"topic-n2"、...等等。

Flink 1.5（FlinkKafkaConsumer09）增加了对动态分区发现的支持

接受subscriptionPattern:link的使用者构造函数。