updataStateByKey算子的使用

updataStateByKeyApp.scala

import org.apache.spark.SparkConf

import org.apache.spark.streaming.{Seconds, StreamingContext}

object updataStateByKeyApp extends App {

  //配置入口点

  val conf = new SparkConf().setAppName(getClass.getSimpleName).setMaster("local[2]")

  val ssc= new StreamingContext(conf, Seconds(1))

  //设置checkpoint的目录

  ssc.checkpoint(".")

  //输入数据流（DStream）

  val lines = ssc.socketTextStream("localhost", 9999)

  //todo...

  val pairs = lines.flatMap(_.split(" ")).map((_,1))

  val runningCounts = pairs.updateStateByKey[Int](updateFunction _)

  //输出打印到控制台

  runningCounts.print()

  //启动StreamingContext，接收数据，然后处理数据

  ssc.start()

  ssc.awaitTermination()

  /**

    * 把当前的数据去更新已有的或者是老的数据

    * @param currentValues  当前的

    * @param preValues 老的

    * @return

    */

  def updateFunction(currentValues: Seq[Int], preValues : Option[Int]): Option[Int] = {

    val current =  currentValues.sum

    val pre = preValues.getOrElse(0)

    Some(current + pre)

  }

}

巴特西

updataStateByKey算子的使用

最新文章

热门文章