spark streaming测试之三有状态的接收数据

2024年3月18日 179次阅读 来源: weixin_33895016

测试思路：

    首先，使用上篇文章的程序一发送网络数据；

    其次，运行spark程序，观察效果。

说明：

    1. 这里使用到了更新函数；

    2. 使用检查点来保证状态。

sparkStreaming

import org.apache.log4j.{LoggerLevel}
import org.apache.spark.streaming.{SecondsStreamingContext}
import org.apache.spark.{SparkContextSparkConf}
import org.apache.spark.streaming.StreamingContext._

object StatefulWordCount {
   def main(args:Array[]){

    Logger.().setLevel(Level.)
    Logger.().setLevel(Level.)

    updateFunc = (values: []state:Option[]) => {
      currentCount = values.foldLeft()(_+_)
      previousCount = state.getOrElse()
      (currentCount + previousCount)
    }

    conf = SparkConf().setAppName().setMaster()
    sc = SparkContext(conf)

    ssc = StreamingContext(sc())
    ssc.checkpoint()

    lines = ssc.socketTextStream(args()args().toInt)
    words = lines.flatMap(_.split())
    wordCounts = words.map(x=>(x))

    stateDstream = wordCounts.updateStateByKey[](updateFunc)
    stateDstream.print()
    ssc.start()
    ssc.awaitTermination()
  }
}

转载于:https://blog.51cto.com/snglw/1656794

    原文作者：weixin_33895016
    原文地址: https://blog.csdn.net/weixin_33895016/article/details/92541300
    本文转自网络文章，转载此文章仅为分享知识，如有侵权，请联系博主进行删除。
点赞