scala - dstream[Double] 到 dstream scala

Question

我正在开发 spark-consumer 应用程序，它正在使用来自 kafka 代理的消息，我想找到将要触发消费者的消息的平均值，最后我想将该平均值存储到 cassandra 中。

val Array(brokers, topics) = args

val sparkConf = new SparkConf().setAppName("MyDirectKafkaWordCount").setMaster("local[2]")
val ssc = new StreamingContext(sparkConf, Seconds(20))

val topicsSet = topics.split(",").toSet
val kafkaParams = Map[String, String]("metadata.broker.list" -> brokers)
val messages = KafkaUtils.createDirectStream[String, String, StringDecoder, StringDecoder](
  ssc, kafkaParams, topicsSet)

val lines = messages.map(_._2)
val count = lines.count()
count.print()
val total = lines.map(JSON.parseFull(_)
    .asInstanceOf[Option[List[Map[String,List[Map[String,Double]]]]]]
    .map(_(1)("MeterData").map(_("mVolts1"))).getOrElse(List())).flatMap(list => list).reduce((x,y) => x+y)
total.print()

val avg = total.reduce((total,count) => total / count )
avg.print()
ssc.start()
ssc.awaitTermination()

在上面的代码中，我得到了总数并准确地计算了我期望的值，但是我无法计算平均值，因为计数是 dstream[long] 并且总数是 dstream[double]。

我认为这条线有一些问题。“val avg = total.reduce((total,count) => total / count)” 任何帮助表示赞赏。

输出：计数： 这是我在流中作为 dstream[Long] 获得的计数输出 总计： 这是我在与 dstream[Double] 相同的流中获得的总输出