我在网上找不到太多关于这个的信息。我想知道是否有可能构建一个flink应用程序，它可以动态地使用与regex模式匹配的所有主题，并将这些主题同步到s3。另外，动态同步的每个主题都有avro消息，flink应用程序将使用confluent的模式注册表。
7rfyedvj1#
真幸运！Flink1.4几天前刚刚发布，这是第一个使用正则表达式提供Kafka主题的版本。根据java文档，以下是如何使用它：
public FlinkKafkaConsumer011(PatternsubscriptionPattern,DeserializationSchema<T> valueDeserializer,Properties props)
为kafka 0.11.x创建新的kafka流媒体源使用者。使用此构造函数可以基于正则表达式模式订阅多个主题。如果启用了分区发现（通过在属性中为flinkkafkaconsumerbase.key\u partition\u discovery\u interval\u millis设置一个非负值），则名称与模式匹配的主题也将在动态创建时订阅。参数：subscriptionpattern—要订阅的主题名称模式的正则表达式。valuedeserializer—用于在kafka的字节消息和flink的对象之间转换的反序列化程序。props—用于配置kafka消费客户端和zookeeper客户端的属性。请注意，运行flink streaming应用程序时，它会以使用consumer配置指定的时间间隔从zookeeper获取主题数据：
FlinkKafkaConsumerBase.KEY_PARTITION_DISCOVERY_INTERVAL_MILLIS
这意味着每个使用者都应该以指定的间隔重新同步包含主题的元数据。默认值为5分钟。因此，添加一个新主题，你应该期望消费者最多在5分钟内开始消费它。您应该使用所需的时间间隔为flink使用者设置此配置。
j0pj023g2#
Flink1.4中添加了使用正则表达式模式订阅Kafka主题。请参阅此处的文档。s3是flink支持的文件系统之一。要将流可靠地一次性传递到文件系统，请使用flink connector文件系统连接器。您可以将flink配置为使用avro，但我不确定与confluent的schema registry的interop状态如何。对于这些和其他主题的搜索，我建议在flink doc页面上进行搜索。例如：https://ci.apache.org/projects/flink/flink-docs-release-1.4/search-results.html?q=schema+registry
2条答案
按热度按时间7rfyedvj1#
真幸运！Flink1.4几天前刚刚发布，这是第一个使用正则表达式提供Kafka主题的版本。根据java文档，以下是如何使用它：
flinkkafkaconsumer011公司
为kafka 0.11.x创建新的kafka流媒体源使用者。使用此构造函数可以基于正则表达式模式订阅多个主题。如果启用了分区发现（通过在属性中为flinkkafkaconsumerbase.key\u partition\u discovery\u interval\u millis设置一个非负值），则名称与模式匹配的主题也将在动态创建时订阅。
参数：
subscriptionpattern—要订阅的主题名称模式的正则表达式。valuedeserializer—用于在kafka的字节消息和flink的对象之间转换的反序列化程序。
props—用于配置kafka消费客户端和zookeeper客户端的属性。
请注意，运行flink streaming应用程序时，它会以使用consumer配置指定的时间间隔从zookeeper获取主题数据：
这意味着每个使用者都应该以指定的间隔重新同步包含主题的元数据。默认值为5分钟。因此，添加一个新主题，你应该期望消费者最多在5分钟内开始消费它。您应该使用所需的时间间隔为flink使用者设置此配置。
j0pj023g2#
Flink1.4中添加了使用正则表达式模式订阅Kafka主题。请参阅此处的文档。
s3是flink支持的文件系统之一。要将流可靠地一次性传递到文件系统，请使用flink connector文件系统连接器。
您可以将flink配置为使用avro，但我不确定与confluent的schema registry的interop状态如何。
对于这些和其他主题的搜索，我建议在flink doc页面上进行搜索。例如：https://ci.apache.org/projects/flink/flink-docs-release-1.4/search-results.html?q=schema+registry