MessageHandler in KafkaUtils010 SparkStreaming

Question

I wanted to group per topic or know from which topic a message comes when applying:

val stream = KafkaUtils.createDirectStream[String, String](
    ssc,
    PreferConsistent, 
    Subscribe[String, String](
      Array(topicConfig.srcTopic),
      kafkaParameters(BOOTSTRAP_SERVERS,"kafka_test_group_id))
    )
  )

However in the latest API kafka010 does not seem to support a message handler as in previous versions. Any idea on how to get the topic?

My goal is to consume from N topics process them (in different ways depending on the topic) and then push it back to another N topics in a 1:1 mapping of the topics:

SrcTopicA--> Process --> DstTopicA
SrcTopicB--> Process --> DstTopicB
SrcTopicC--> Process --> DstTopicC

But there are some attributes that need to be shared (that change a lot so there is no possibility of using a broadcast variable). So all the topics need to be consumed in the same spark job.

Answer 1

You can filter the stream using topic like this:

stream.filter(cr => cr.topic)

Answer 2

When you use createDirectStream in 0.10 you get back a ConsumerRecord. This record has a topic value. You can create a tuple of the topic and value:

val stream: InputDStream[ConsumerRecord[String, String]] = 
  KafkaUtils.createDirectStream[String, String](
    streamingContext,
    PreferConsistent,
    Subscribe[String, String](topics, kafkaParams)
  )

val res: DStream[(String, String)] = stream.map(record => (record.topic(), record.value()))