sparkStreaming带状态更新（scala）

港控/mmm° 2022-07-12 23:39 161阅读 0赞

带状态的更新是使用的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()
      }
    }

发表评论取消回复

表情：

评论列表（有 0 条评论，161人围观）

还没有评论，来说两句吧...

相关阅读

相关 SparkStreaming (一) --------- SparkStreaming 概述

目录一、Spark Streaming 是什么 ? 二、Spark Streaming 的特点二、Spark Streaming 架构

小鱼儿/ 2024年04月07日 13:12/ 0 赞/ 117 阅读

相关 Thymeleaf：更新状态（六）

需求一份未读奏折，点击查看后，需要从未读变成已读。修改WorkServlet中showMemorialsDetail方法 protected void

矫情吗；*/ 2024年03月22日 14:35/ 0 赞/ 85 阅读

相关 SparkStreaming-有状态的数据转换

DStream上的原语与RDD的类似，分为Transformations（转换）和Output Operations（输出）两种，此外转换操作中还有一些比较特殊的原语，如：up

偏执的太偏执、/ 2023年02月19日 13:29/ 0 赞/ 43 阅读

相关 SparkStreaming概述

SparkStreaming概述 1.1 Spark Streaming是什么 ![在这里插入图片描述][20210126223839932.png] ![在这

忘是亡心i/ 2023年01月11日 13:00/ 0 赞/ 209 阅读

相关 sparkStreaming带状态更新（scala）

带状态的更新是使用的updateStateByKey方法，里面传入一个函数，函数小自己写，注意需要设置checkpoint package sparkStreamin

港控/mmm°/ 2022年07月12日 23:39/ 0 赞/ 162 阅读

相关 Spark--sparkStreaming

概述 Spark Streaming是核心Spark API的扩展，可实现可扩展、高吞吐量、可容错的实时数据流处理。数据可以从诸如Kafka，Flume，Kines

左手的ㄟ右手/ 2022年05月17日 07:10/ 0 赞/ 300 阅读

相关 SparkStreaming DStream

DStream就是离散流，它代表了一个持续不断的数据流，DStream的内部，其实是一系列不断产生的RDD,RDD是Spark Core的核心抽象，DStream中的每个RDD

谁借莪１个温暖的怀抱￠/ 2022年05月16日 11:15/ 0 赞/ 274 阅读

相关 sparkStreaming 报缺scala包的问题

![20190418133125431.png][] ![20190418133154710.png][] ![20190418133213671.png][]

╰半夏微凉°/ 2022年04月24日 07:12/ 0 赞/ 128 阅读

相关 SparkStreaming例题

取材自官网http://Spark.apache.org 案例1：和集群搭配使用 package SparkStreaming import ja

太过爱你忘了你带给我的痛/ 2022年04月02日 07:26/ 0 赞/ 277 阅读

相关 SparkStreaming

SparkStreaming 一、SparkStreaming简介 SparkStreaming是流式处理框架，是Spark API的扩展，支持可扩展、高吞吐量

痛定思痛。/ 2021年11月23日 14:20/ 0 赞/ 413 阅读