使用Flink动态消费和接收Kafka主题

Question

我无法在网上找到有关此内容的更多信息。我想知道是否可以构建一个Flink应用程序，它可以动态地使用与正则表达式模式匹配的所有主题并将这些主题同步到s3。此外，动态同步的每个主题都会有Avro消息，而Flink应用程序将使用Confluent的Schema Registry。

Answer 1

好幸运！ Flink 1.4几天前刚刚发布，这是第一个使用REGEX提供消费Kafka主题的版本。根据java文档，这里是如何使用它：

FlinkKafkaConsumer011

public FlinkKafkaConsumer011(PatternsubscriptionPattern,DeserializationSchema<T> valueDeserializer,Properties props)

为Kafka 0.11.x创建一个新的Kafka流媒体源消费者。使用此构造函数可基于正则表达式模式订阅多个主题。如果启用了分区发现（通过在属性中为FlinkKafkaConsumerBase.KEY_PARTITION_DISCOVERY_INTERVAL_MILLIS设置非负值），则还将订阅名称与模式匹配的主题，因为它们是即时创建的。

参数：

subscriptionPattern - 要订阅的主题名称模式的正则表达式。 valueDeserializer - 用于在Kafka的字节消息和Flink的对象之间进行转换的de / serializer。

props - 用于配置Kafka使用者客户端和ZooKeeper客户端的属性。

只需注意运行Flink流应用程序，它会使用使用者配置指定的间隔从Zookeeper获取主题数据：

FlinkKafkaConsumerBase.KEY_PARTITION_DISCOVERY_INTERVAL_MILLIS

这意味着每个消费者都应该按照指定的时间间隔重新同步元数据，包括主题。默认值为5分钟。因此，添加一个新主题，您应该期望消费者最多在5分钟内开始消费它。您应该使用所需的时间间隔为Flink使用者设置此配置。

Answer 2

在Flink 1.4中添加了使用正则表达式模式订阅Kafka主题。见documentation here。

S3是Flink支持的文件系统之一。要将流可靠，准确地一次传送到文件系统，请使用flink-connector-filesystem connector。

您可以将Flink配置为使用Avro，但我不确定与Confluent的架构注册表互操作的状态。

要搜索这些主题和其他主题，我建议您在Flink文档页面上进行搜索。例如：https://ci.apache.org/projects/flink/flink-docs-release-1.4/search-results.html?q=schema+registry