Kafka的Spark流：如何从Kafka消费者数据流中获取主题名称？

Question

我在Scala中设置了Spark Kafka Consumer，它接收来自多个主题的消息：

val properties = readProperties()
val streamConf = new SparkConf().setMaster("local[*]").setAppName("Kafka-Stream")
val ssc = new StreamingContext(streamConf, Seconds(10))

val kafkaParams = Map("metadata.broker.list" ->  properties.getProperty("broker_connection_str"), 
                      "zookeeper.connect"    ->  properties.getProperty("zookeeper_connection_str"), 
                      "group.id"             ->  properties.getProperty("group_id"), 
                      "auto.offset.reset"    ->  properties.getProperty("offset_reset")
                    )

// Kafka integration with receiver 
val msgStream = KafkaUtils.createStream[Array[Byte], String, DefaultDecoder, StringDecoder](
  ssc, kafkaParams, Map(properties.getProperty("topic1") -> 1,
                      properties.getProperty("topic2") -> 2,
                      properties.getProperty("topic3") -> 3),
                      StorageLevel.MEMORY_ONLY_SER).map(_._2)

我需要为每个主题的消息（将采用JSON格式）开发相应的操作代码。

我提到了以下问题，但其中的答案对我没有帮助：

从spark中的Kafka消息获取主题

那么，在接收到的DStream上是否有任何方法可用于获取主题名称以及消息以确定应该采取什么行动？

对此任何帮助都将不胜感激。谢谢你。

Answer 1

请参阅下面的代码。

您可以通过foreachRDD获取主题名称和消息，在DStream上进行地图操作。

msgStream.foreachRDD(rdd => {
      val pairRdd = rdd.map(i => (i.topic(), i.value()))
})

下面的代码是我正在使用的createDirectStream的示例源代码。

val ssc = new StreamingContext(configLoader.sparkConfig, Seconds(conf.getInt(Conf.KAFKA_PULL_INTERVAL)))
val kafkaParams = Map[String, Object](
  "bootstrap.servers" -> conf.getString(Conf.KAFKA_BOOTSTRAP_SERVERS),
  "key.deserializer" -> classOf[StringDeserializer],
  "value.deserializer" -> classOf[StringDeserializer],
  "group.id" -> conf.getString(Conf.KAFKA_CONSUMER_GID),
  "auto.offset.reset" -> conf.getString(Conf.KAFKA_AUTO_OFFSET_RESET),
  "enable.auto.commit" -> (false: java.lang.Boolean)
)
val topics: Array[String] = conf.getString(Conf.KAFKA_TOPICS).split(",")
val stream = KafkaUtils.createDirectStream[String, String](
  ssc,
  PreferConsistent,
  Subscribe[String, String](topics, kafkaParams)
)