仅处理 Kafka 流中的第一条消息

Question

在 Spark 中，我从 Kafka 创建了一个批处理时间为 5 秒的流。在此期间可能会收到许多消息，我想单独处理每条消息，但按照我目前的逻辑，似乎只处理每批的第一条消息。

val stream = KafkaUtils.createDirectStream[String, String, StringDecoder, StringDecoder](ssc, params, topics)

val messages = stream.map((x$2) => x$2._2)

messages.foreachRDD { rdd =>
    if(!rdd.isEmpty) {
        val message = rdd.map(parse)
        println(message.collect())
    }
}

parse 函数只是将 Json 消息中的相关字段提取到一个元组中。

我可以深入了解分区并以这种方式单独处理每条消息：

messages.foreachRDD { rdd =>
    if(!rdd.isEmpty) {
        rdd.foreachPartition { partition =>
            partition.foreach{msg =>
                val message = parse(msg)
                println(message)
            }
        }
    }
}

但我确信有一种方法可以保持在 RDD 级别。在第一个示例中我做错了什么？

我正在使用 spark 2.0.0、scala 2.11.8 和 spark streaming kafka 0.8。

Answer 1

这是示例流应用程序，它将批处理的每条消息转换为每个循环内部的大写字母并打印它们。试试这个示例应用程序，然后重新检查您的应用程序。希望这会有所帮助。

object SparkKafkaStreaming {

def main(args: Array[String]) {

//Broker and topic
val brokers = "localhost:9092"
val topic = "myTopic"

//Create context with 5 second batch interval
val sparkConf = new SparkConf().setAppName("SparkKafkaStreaming").setMaster("local[2]")
val ssc = new StreamingContext(sparkConf, Seconds(5))

//Create direct kafka stream with brokers and topics
val topicsSet = Set[String](topic)
val kafkaParams = Map[String, String]("metadata.broker.list" -> brokers)
val msgStream = KafkaUtils.createDirectStream[String, String, StringDecoder, StringDecoder](ssc, kafkaParams, topicsSet)

//Message
val msg = msgStream.map(_._2)    
msg.print()

//For each
msg.foreachRDD { rdd =>
  if (!rdd.isEmpty) {
     println("-----Convert Message to UpperCase-----")
    //convert messages to upper case
    rdd.map { x => x.toUpperCase() }.collect().foreach(println)
  } else {
    println("No Message Received")
  }
}

//Start the computation
ssc.start()
ssc.awaitTermination()
  }
}