spark streamingテストの4設定ウィンドウサイズ受信データ
テストの考え方:
まず、ネットワークデータ送信プログラムを用いてデータを送信する.
その後、sparkプログラムを実行します.
効果を観察する.
説明:
1. ここでもチェックポイントディレクトリを設定する必要があります
2. ここには4つのパラメータがあります.
最初の2つは、それぞれリスニングされたポートと、数ミリ秒ごとにデータを受信します.
3番目のパラメータは、受信数ミリ秒前のデータです.(詳細はwindowの具体的な意味を参照)
4番目のパラメータは、数ミリ秒ごとにデータを受信します.
まず、ネットワークデータ送信プログラムを用いてデータを送信する.
その後、sparkプログラムを実行します.
効果を観察する.
説明:
1. ここでもチェックポイントディレクトリを設定する必要があります
2. ここには4つのパラメータがあります.
最初の2つは、それぞれリスニングされたポートと、数ミリ秒ごとにデータを受信します.
3番目のパラメータは、受信数ミリ秒前のデータです.(詳細はwindowの具体的な意味を参照)
4番目のパラメータは、数ミリ秒ごとにデータを受信します.
sparkStreaming
import org.apache.log4j.{LoggerLevel}
import org.apache.spark.storage.StorageLevel
import org.apache.spark.streaming.{SecondsStreamingContext}
import org.apache.spark.{SparkContextSparkConf}
import org.apache.spark.streaming.StreamingContext._
WindowWordCount {
def main(args: Array[]){
Logger.().setLevel(Level.)
Logger.().setLevel(Level.)
conf = SparkConf().setAppName().setMaster()
sc = SparkContext(conf)
ssc = StreamingContext(sc())
ssc.checkpoint()
val lines = ssc.socketTextStream(args(0),args(1).toInt,
StorageLevel.MEMORY_ONLY_SER)
words = lines.flatMap(_.split())
//windows operator
val wordCounts = words.map(x=>(x,1)).reduceByKeyAndWindow((a:Int,b:Int)=>(a+b),
Seconds(args(2).toInt),Seconds(args(3).toInt))
wordCounts.print()
ssc.start()
ssc.awaitTermination()
}
}