频道栏目
首页 > 网络 > 云计算 > 正文
sparkStreaming带状态更新(scala)
2016-12-21 09:29:00         来源:韩利鹏的博客  
收藏   我要投稿

sparkStreaming带状态更新(scala):带状态的更新是使用的updateStateByKey方法,里面传入一个函数,函数小自己写,注意需要设置checkpoint。

package sparkStreaming__

import org.apache.spark.streaming.kafka.KafkaUtils
import org.apache.spark.streaming.{Seconds, StreamingContext}
import org.apache.spark.{SparkConf, SparkContext}

/**
  * Created by root on 2016/12/14.
  * 需要设置checkpoint
  * 有状态的计算
  */
class UpdataByKey {

}
object UpdataByKey{
    //自定义函数进行带状态更新
  def addFunc (currValue:Seq[Int],point:Option[Int])={
    Some(currValue.sum+point.getOrElse(0));
  }

  def main(args: Array[String]): Unit = {
    val conf = new SparkConf().setAppName("UpdataByKey").setMaster("local[*]")
    val sc = new SparkContext(conf)
    val ssc = new StreamingContext(sc,Seconds(10))
    val topics = "xiaopeng";
    val topicMap = topics.split(",").map((_,2)).toMap
    val lines = KafkaUtils.createStream(ssc,"192.168.10.219:2181","han",topicMap)
    val words = lines.flatMap(line =>line._2.split(" ")).map(word =>(word,1))
    words.updateStateByKey[Int](addFunc _)
    words.print()
    ssc.start()
    ssc.awaitTermination()
  }
}
点击复制链接 与好友分享!回本站首页
上一篇:关于opentack的(3)novaapi简析
下一篇:sparksql各种数据源
相关文章
图文推荐
点击排行

关于我们 | 联系我们 | 广告服务 | 投资合作 | 版权申明 | 在线帮助 | 网站地图 | 作品发布 | Vip技术培训 | 举报中心

版权所有: 红黑联盟--致力于做实用的IT技术学习网站