Kyle Hamlin created FLINK-8380:
----------------------------------
Summary: Dynamic BucketingSink paths based on ingested Kafka topics
Key: FLINK-8380
URL:
https://issues.apache.org/jira/browse/FLINK-8380 Project: Flink
Issue Type: Improvement
Components: DataStream API
Affects Versions: 1.4.0
Reporter: Kyle Hamlin
Flink 1.4 released a feature that allows Kafka consumers to dynamically ingest topics based on a regex pattern. If a user wanted to use Flink as a simple (no transformations) but dynamic (auto topic discovery & auto output path generation) data persister they would currently only have half the tools to do so. I believe it would be a beneficial feature to allow users to not only define automatic topic discovery but also a way to dynamically incorporate those topics into a BucketingSink output path. For example:
If I had three Kafka topics
{code:java}
select-topic-1
ignore-topic-1
select-topic-2
{code}
And my Kafka consumers regex only selected two topics
{code:java}
val consumer = new FlinkKafkaConsumer010[GenericRecord](Pattern.compile("select-.*?"), new MyDeserializer(), props)
{code}
Then the selected topics would appended to the beginning of the BucketingSink output path and any Bucketers partitions would follow
{code:java}
val sink = new BucketingSink[GenericRecord]("s3://my-bucket/")
sink.setBucketer(new DateTimeBucketer[GenericRecord]("yyyyMMdd"))
{code}
The resulting output paths would be
{code:java}
s3://my-bucket/selected-topic1/yyyyMMdd/
s3://my-bucket/selected-topic2/yyyyMMdd/
{code}
As new topics are discovered via the regex pattern (while the app is running) the set of BucketingSink output paths would grow.
--
This message was sent by Atlassian JIRA
(v6.4.14#64029)
Locked
