flink-user mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From "Tzu-Li (Gordon) Tai" <tzuli...@apache.org>
Subject Re: how to get topic names in SinkFunction when using FlinkKafkaConsumer010 with multiple topics
Date Mon, 17 Jul 2017 05:40:10 GMT
Hi,

Here’s an example:

DataStream<String> inputStream = …;

inputStream.addSink(new FlinkKafkaProducer09<>(
    “defaultTopic”, new CustomKeyedSerializationSchema(), props));

Code for CustomKeyedSerializationSchema:
public class CustomKeyedSerializationSchema implements KeyedDeserializationSchema<String>
{
    byte[] getKeyBytes(…) {…}
    byte[] getValueBytes(…) {…}
    String getTargetTopic() {
      ...
    }
}

For the above code, by default records will always be sent to the “defaultTopic” topic.
However, for each record, `getTargetTopic` will also be called. Whatever is returned from
there will override “defaultTopic”.
You can place your switch there.

Cheers,
Gordon

On 6 July 2017 at 11:43:55 PM, Richard Xin (richardxin168@yahoo.com) wrote:

Thanks,
I'm not sure I understand this, what I need is to have single a process subscribing multiple
kafka topics, and have a switch clause for different topics in my SinkFunction, did you I
need to change the way how the kafka producer to produce the message? 
Any pointer to code samples will be appreciated. 
Thanks Again
Richard


On Wednesday, July 5, 2017, 10:25:59 PM PDT, Tzu-Li (Gordon) Tai <tzulitai@apache.org>
wrote:


Hi Richard,

Producing to multiple topics is treated a bit differently in the Flink Kafka producer.
You need to set a single default target topic, and in `KeyedSerializationSchema#getTargetTopic()`
you can override the default topic with whatever is returned. The `getTargetTopic` method
is invoked for each record.

Cheers,
Gordon


On 6 July 2017 at 9:09:29 AM, Richard Xin (richardxin168@yahoo.com) wrote:

when using FlinkKafkaConsumer010 to subscribing multiple topics as 
List<String> topics = Arrays.asList("test1","test2");

DataStream<String> stream = env.addSource(new FlinkKafkaConsumer010<>(topics, new SimpleStringSchema(), properties));

How do I get topic names in my SinkFunction? i.e. stream.addSink()

Thanks,
Richard
Mime
View raw message