Lukasz, I appreciate the quick response and filing the JIRA ticket. Thanks for the suggestion, unfortunately, I don't have a fixed number of topics. Still, we'll probably use your approach for a limited number of topics until the functionality is added, thank you!
Thanks, Dmitry On Fri, Oct 19, 2018 at 2:53 PM Lukasz Cwik <[email protected]> wrote: > If there are a fixed number of topics, you could partition your write by > structuring your pipeline as such: > ParDo(PartitionByTopic) ----> KafkaIO.write(topicA) > \---> KafkaIO.write(topicB) > \---> KafkaIO.write(...) > > There is no support currently for writing to Kafka dynamically based upon > a destination that is part of the data. > I filed https://issues.apache.org/jira/browse/BEAM-5798 for the issue. > > On Fri, Oct 19, 2018 at 2:05 PM [email protected] <[email protected]> > wrote: > >> Hi guys!! >> >> I'm trying to find a way to write to a Kafka topic using KafkaIO.write() >> But I need to be able to get topic name dynamically based on the data >> received. For example, I would like to send data for one tenant to topic >> "data_feed_1" and for another tenant to "topic data_feed_999". >> I'm coming from Flink where it's possible via >> KeyedSerializationSchema.getTargetTopic(). >> Is there anything similar in KafkaIO? >> >> Thanks, >> Dmitry >> > -- -- Dmitry
