在不同的路径下将几个 Kafka 主题下沉到 s3

Question

随着 Flink 1.4 的发布，FlinkKafkaConsumer011 具有传递正则表达式以供阅读主题的功能——顺便说一句，这是一个 rad 功能。现在我想知道是否有一种方法可以让单个 Flink 应用程序使用 BucketingSink 来接收每个主题（Avro 消息）以分隔 s3 中的位置。例如：

s3://bucket/topic_1
s3://bucket/topic_2
s3://bucket/topic_3
.
.
.
s3://bucket/topic_n

非常感谢任何关于如何实现这一点的指针！

Answer 1

如果您的根存储桶不同，并且您“只是”希望每个主题有不同的路径，您可以：

1. 从 Kafka 反序列化消息时，从元数据中获取主题并将其存储在您的元素中
2. 通过完整处理将其传递给您的元素
3. 提供 BasePathBucketer 的子类，它重新定义了 getBucketPath 方法，通过在传递的元素中查询主题来返回带有主题的路径。