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
>
