2

我有这个简单的 Kafka Stream

val messages = KafkaUtils.createDirectStream[String, String, StringDecoder, StringDecoder](ssc, kafkaParams, topicsSet)

// Each Kafka message is a flight
val flights = messages.map(_._2)

flights.foreachRDD( rdd => {
  println("--- New RDD with " + rdd.partitions.length + " partitions and " + rdd.count() + " flight records");
  rdd.map { flight => {        
    val flightRows = FlightParser.parse(flight)
    println ("Parsed num rows: " + flightRows)
    }
  }          
})

ssc.start()
ssc.awaitTermination()

Kafka 有消息，Spark Streaming 能够将它们作为 RDD 获取。但是我的代码中的第二个 println 不打印任何内容。我在本地 [2] 模式下运行时查看了驱动程序控制台日志，在纱线客户端模式下运行时检查了纱线日志。

我错过了什么？

以下代码代替 rdd.map，在 spark 驱动程序控制台中可以很好地打印：

for(flight <- rdd.collect().toArray) {
     val flightRows = FlightParser.parse(flight)
     println ("Parsed num rows: " + flightRows)
}

但我担心这个飞行对象的处理可能发生在火花驱动程序项目中，而不是执行程序中。如果我错了，请纠正我。

谢谢

4

1 回答 1

2

rdd.map是一个惰性转换。除非对该 RDD 调用操作，否则它不会实现。
在这种特定情况下，我们可以使用rdd.foreach 哪个是 RDD 上最通用的操作之一，使我们能够访问 RDD 中的每个元素。

flights.foreachRDD{ rdd => 
    rdd.foreach { flight =>        
        val flightRows = FlightParser.parse(flight)
        println ("Parsed num rows: " + flightRows) // prints on the stdout of each executor independently
    }
}

鉴于这个 RDD 动作是在 executor 中执行的，我们会在 executor 的 STDOUT 中找到 println 输出。

如果您想打印驱动程序上的数据，您可以collect在DStream.foreachRDD闭包中打印 RDD 的数据。

flights.foreachRDD{ rdd => 
  val allFlights = rdd.collect() 
  println(allFlights.mkString("\n")) // prints to the stdout of the driver
}
于 2016-04-12T10:29:05.097 回答