object StreamingWordCount {

  def main(args: Array[String]): Unit = {
    if (args.length < 2) {
      System.err.println("Usage: NetworkWordCount <hostname> <port>")
      System.exit(1)
    }

    val sparkConf = new SparkConf().setAppName("NetworkWordCount")
    val ssc = new StreamingContext(sparkConf, Seconds(1))              

    val lines = ssc.socketTextStream(                                  
      hostname = args(0),
      port = args(1).toInt,
      storageLevel = StorageLevel.MEMORY_AND_DISK_SER)                 

    val words = lines.flatMap(_.split(" "))
    val wordCounts = words.map(x => (x, 1)).reduceByKey(_ + _)

    wordCounts.print()                                                 

    ssc.start()                                                        
    ssc.awaitTermination()                                             
  }
}

val conf = new SparkConf().setAppName(appName).setMaster(master)
val ssc = new StreamingContext(conf, Seconds(1))

val textRDD = sc.textFile(...)
val splitRDD = textRDD.flatMap(_.split(" "))
val tupleRDD = splitRDD.map((_, 1))
val reduceRDD = tupleRDD.reduceByKey(_ + _)

val lines: DStream[String] = ssc.socketTextStream(...)
val words: DStream[String] = lines.flatMap(_.split(" "))
val wordCounts: DStream[(String, Int)] = words.map(x => (x, 1)).reduceByKey(_ + _)

val sparkConf = new SparkConf().setAppName("NetworkWordCount").setMaster("local[6]")
val sc = new SparkContext(sparkConf)
sc.setLogLevel("ERROR")
val ssc = new StreamingContext(sc, Seconds(1))

val lines: DStream[String] = ssc.socketTextStream(
  hostname = "localhost",
  port = "9999".toInt,
  storageLevel = StorageLevel.MEMORY_AND_DISK_SER)

val words = lines.flatMap(_.split(" ")).map(x => (x, 1))

// 使用 updateStateByKey 必须设置 Checkpoint 目录
ssc.checkpoint("checkpoint")

// updateStateByKey 的函数
def updateFunc(newValue: Seq[Int], runningValue: Option[Int]) = {
  // newValue 之所以是一个 Seq, 是因为它是某一个 Batch 的某个 Key 的全部 Value
  val currentBatchSum = newValue.sum
  val state = runningValue.getOrElse(0)
  // 返回的这个 Some(count) 会再次进入 Checkpoint 中当作状态存储
  Some(currentBatchSum + state)
}

// 调用
val wordCounts = words.updateStateByKey[Int](updateFunc)

wordCounts.print()

ssc.start()
ssc.awaitTermination()

val sparkConf = new SparkConf().setAppName("NetworkWordCount").setMaster("local[6]")
val sc = new SparkContext(sparkConf)
sc.setLogLevel("ERROR")
val ssc = new StreamingContext(sc, Seconds(1))

val lines: DStream[String] = ssc.socketTextStream(
  hostname = "localhost",
  port = 9999,
  storageLevel = StorageLevel.MEMORY_AND_DISK_SER)

val words = lines.flatMap(_.split(" ")).map(x => (x, 1))

// 通过 window 操作, 会将流分为多个窗口
val wordsWindow = words.window(Seconds(30), Seconds(10))
// 此时是针对于窗口求聚合
val wordCounts = wordsWindow.reduceByKey((newValue, runningValue) => newValue + runningValue)

wordCounts.print()

ssc.start()
ssc.awaitTermination()

val sparkConf = new SparkConf().setAppName("NetworkWordCount").setMaster("local[6]")
val sc = new SparkContext(sparkConf)
sc.setLogLevel("ERROR")
val ssc = new StreamingContext(sc, Seconds(1))

val lines: DStream[String] = ssc.socketTextStream(
  hostname = "localhost",
  port = 9999,
  storageLevel = StorageLevel.MEMORY_AND_DISK_SER)

val words = lines.flatMap(_.split(" ")).map(x => (x, 1))

// 开启窗口并自动进行 reduceByKey 的聚合
val wordCounts = words.reduceByKeyAndWindow(
  reduceFunc = (n, r) => n + r,
  windowDuration = Seconds(30),
  slideDuration = Seconds(10))

wordCounts.print()

ssc.start()
ssc.awaitTermination()