6

以下は、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))

誰が問題が何であるか教えてもらえますか? ありがとう！

4

1 に答える 1

10

これは非常に簡単に修正できます。型を明示するだけです。
val wordCounts = words.map((_, 1)).reduceByKeyAndWindow((a:Int,b:Int)=>a+b, Seconds(30), Seconds(10))

この場合、scala が型を推測できない理由は、この回答で説明されています

于 2014-07-22T17:37:26.787 に答える