Scala调用updateStateByKey比Java简单太多

使用Scala的写法调用有状态转换算子updateStateByKey方法，直接传函数

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


object StreamingWordCount {
  def main(args: Array[String]): Unit = {

    // 设置将来访问 hdfs 的使用的用户名, 否则会出现全选不够
    System.setProperty("HADOOP_USER_NAME", "chen")

    // 1. 创建SparkStreaming的入口对象: StreamingContext
    //Seconds(5): 表示事件间隔5秒
    val conf = new SparkConf().setAppName("StreamingWordCount").setMaster("local[*]")
    val ssc = new StreamingContext(conf, Seconds(5))

    // 2.设置检查点: 使用updateStateByKey必须设置检查点，checkpoint文件保存在当前目录下
    ssc.checkpoint("checkpoint")

    // 3. Socket方式创建一个DStream
    val lines: ReceiverInputDStream[String] = ssc.socketTextStream("hadoop103", 9999)

    // 4. 切分成一个个的单词
    val words: DStream[String] = lines.flatMap(_.split("\W+"))

    // 5. 将单词拼成元组
    val wordAndOne: DStream[(String, Int)] = words.map((_, 1))

    /*
    6.求新的阶段总数和上一个阶段进行求和操作
      参数1(seq): 在当前阶段 一个新的key对应的value组成的序列  在我们这个案例中是: 1,1,1,1...
      参数2(option): 上一个阶段 这个key对应的value
      (0 /: seq)(_ + _)等同于seq.foldleft(0)(_ + _)使用函数的左折叠求和，此处也可以用seq.sum求和
    */
    val stateDS: DStream[(String, Int)] = wordAndOne.updateStateByKey[Int](
      (seq: Seq[Int], option: Option[Int]) => Some((0 /: seq)(_ + _) + option.getOrElse(0))
    )

    // 7. 结果打印在控制台
    stateDS.print

    // 8. 启动任务开始计算
    ssc.start()
    ssc.awaitTermination()
  }
}