Kafka的Spark流：如何从Kafka consumer DStream获取主题名称？

Question

我已经设置了多个主题接收消息的火花卡夫卡消费者在斯卡拉：

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格式）。

我提到了以下问题，但在它的回答对我没有帮助：

get topic from Kafka message in spark

那么，有没有可以用来获取沿着主题名称接收到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) 
)