以下是获取窗口大小为 30 秒、幻灯片大小为 10 秒的字数统计的简单代码。
import org.apache.spark.SparkConf
import org.apache.spark.streaming._
import org.apache.spark.streaming.StreamingContext._
import org.apache.spark.api.java.function._
import org.apache.spark.streaming.api._
import org.apache.spark.storage.StorageLevel
val ssc = new StreamingContext(sc, Seconds(5))
// read from text file
val lines0 = ssc.textFileStream("test")
val words0 = lines0.flatMap(_.split(" "))
// read from socket
val lines1 = ssc.socketTextStream("localhost", 9999, StorageLevel.MEMORY_AND_DISK_SER)
val words1 = lines1.flatMap(_.split(" "))
val words = words0.union(words1)
val wordCounts = words.map((_, 1)).reduceByKeyAndWindow(_ + _, Seconds(30), Seconds(10))
wordCounts.print()
ssc.checkpoint(".")
ssc.start()
ssc.awaitTermination()
但是,我从这一行收到错误:
val wordCounts = words.map((_, 1)).reduceByKeyAndWindow(_ + _, Seconds(30), Seconds(10))
。特别是,从_ + _
。错误是
51: error: missing parameter type for expanded function ((x$2, x$3) => x$2.$plus(x$3))
有人能告诉我问题是什么吗?谢谢!
这非常容易修复,只需明确类型即可。
val wordCounts = words.map((_, 1)).reduceByKeyAndWindow((a:Int,b:Int)=>a+b, Seconds(30), Seconds(10))
在这种情况下 scala 无法推断类型的原因解释如下这个答案 https://stackoverflow.com/questions/9118264/multiple-parameter-closure-argument-type-not-inferred
本文内容由网友自发贡献,版权归原作者所有,本站不承担相应法律责任。如您发现有涉嫌抄袭侵权的内容,请联系:hwhale#tublm.com(使用前将#替换为@)