技術標籤：linux

package com.ws.sparkstreaming

import org.apache.spark.streaming.dstream.DStream
import org.apache.spark.streaming.{Seconds, StreamingContext}
import org.apache.spark.{SparkConf, SparkContext}

object WordCount {
  def main(args: Array[String]): Unit = {
    // 建立conf
    val conf = new SparkConf().setAppName(this.getClass.getSimpleName).setMaster("local[*]")
    // 建立sparkContext
    val sc = new SparkContext(conf)
    // 建立streamingContext，他是對sparkContext的包裝，後邊的Seconds(2)只得是批處理的時間間隔
    val scc = new StreamingContext(sc,Seconds(2))
    // 監聽socket資料
    val stream: DStream[String] = scc.socketTextStream("dream1", 8888)
    // rdd操作，但是不是spark的rdd，不過最後還是用的spark的rdd
    val value: DStream[(String, Int)] = stream.flatMap(_.split(" ")).map((_, 1)).reduceByKey(_ + _)
    // streaming的列印方法。
    value.print()
    // 必須要啟動才行
    scc.start()
    // 不退出
    scc.awaitTermination()
  }
}
 

linux 安裝nc客戶端 yum install nc -y

nc -lk 8888