spark streaming检查点使用

import org.apache.spark._
import org.apache.spark.streaming._ /**
* Created by code-pc on 16/3/14.
*/
object Pi { def functionToCreateContext():StreamingContext={ def updateStateFunction(newValues: Seq[Int], runningCount: Option[Int]): Option[Int] = {
Some(runningCount.getOrElse(0) + newValues.sum)
} val conf = new SparkConf().setMaster("local[5]").setAppName("AndrzejApp")
val ssc = new StreamingContext(conf, Seconds(4))
ssc.checkpoint("/tmp") val line = ssc.socketTextStream("127.0.0.1", 9997)
val words=line.flatMap(_.split(" ")) val pairs=words.map((_,1)) val retDS = pairs.updateStateByKey[Int](updateStateFunction _) //retDS.repartition(1).saveAsTextFiles("/tmp/out")
retDS.print() ssc
} def main(args: Array[String]) { val context=StreamingContext.getOrCreate("/tmp",functionToCreateContext) context.start()
context.awaitTermination()
}
}

tcp: nc -lk 9997

If the checkpointDirectory exists, then the context will be recreated from the checkpoint data.

上一篇:php 网站301重定向设置代码实战案例


下一篇:PAT 1002. 写出这个数 (20)