最新的购物网站 开,网站改版方案怎么写,asp网站设置,企业网站设计过程中一、上下文
《Spark-Streaming初识》中的NetworkWordCount示例只能统计每个微批下的单词的数量#xff0c;那么如何才能统计从开始加载数据到当下的所有数量呢#xff1f;下面我们就来通过官方例子学习下Spark-Streaming有状态计算。
二、官方例子
所属包#xff1a;org.…一、上下文
《Spark-Streaming初识》中的NetworkWordCount示例只能统计每个微批下的单词的数量那么如何才能统计从开始加载数据到当下的所有数量呢下面我们就来通过官方例子学习下Spark-Streaming有状态计算。
二、官方例子
所属包org.apache.spark.examples.streaming
object StatefulNetworkWordCount {def main(args: Array[String]): Unit {if (args.length 2) {System.err.println(Usage: StatefulNetworkWordCount hostname port)System.exit(1)}StreamingExamples.setStreamingLogLevels()val sparkConf new SparkConf().setAppName(StatefulNetworkWordCount)//创建微批为 1 秒的上下文val ssc new StreamingContext(sparkConf, Seconds(1))//指定 checkpoint 目录ssc.checkpoint(.)// 用一个 List 初始化一个 RDDval initialRDD ssc.sparkContext.parallelize(List((hello, 1), (world, 1)))// 在目标ip:port上创建一个ReceiverInputDStream并对分隔测试的输入流中的单词进行计数例如由nc生成val lines ssc.socketTextStream(args(0), args(1).toInt)val words lines.flatMap(_.split( ))val wordDstream words.map(x (x, 1))// 使用mapWithState更新累积计数这将给出一个由状态组成的DStream即单词的累积计数val mappingFunc (word: String, one: Option[Int], state: State[Int]) {val sum one.getOrElse(0) state.getOption.getOrElse(0)val output (word, sum)state.update(sum)output}val stateDstream wordDstream.mapWithState(StateSpec.function(mappingFunc).initialState(initialRDD))stateDstream.print()ssc.start()ssc.awaitTermination()}
}
三、分析
1、构建SparkConf
它是Spark应用程序的配置用于设置Spark的各种参数。支持链式设置
new SparkConf().setMaster(local).setAppName(My app) 一旦SparkConf对象传递给Spark用户就不能再对其进行修改。Spark不支持在运行时修改配置
2、构建StreamingContext
它是Spark Streaming功能的主要入口点且提供了从各种输入源创建[[org.apache.spark.streaming.dstream.DStream]] 的方法。
创建和转换DStreams后可以分别使用start()、stop()启动和停止流计算awaitTermination()允许当前线程通过stop()或异常等待上下文的终止。
3、设置checkpoint
StreamingContext最终还是通过SparkContext来设置checkpoint但其实都是为各自的checkpointDir设置checkpoint路径在有状态计算中checkpoint是必须的。
所谓有状态计算就必须要把历史状态给存储下来spark中使用使用checkpoint来实现这个存储每个微批的数据的计算都要更新到历史状态中。
class SparkContext(config: SparkConf) extends Logging {private[spark] var checkpointDir: Option[String] None}
class StreamingContext private[streaming] (_sc: SparkContext,_cp: Checkpoint,_batchDur: Duration) extends Logging {private[streaming] var checkpointDir: String {if (isCheckpointPresent) {sc.setCheckpointDir(_cp.checkpointDir)_cp.checkpointDir} else {null}}}
4、初始化一个RDD
为什么要初始化一个RDD呢我们看看下面是如何用到的。
5、创建一个ReceiverInputDStream
这里是从TCP源hostname:port创建输入流。使用TCP套接字接收数据并使用给定的转换器将接收字节解释为对象
6、处理单词
从源码中可以看出会把这样的文本 hadoop spark flink kafka hadoop spark-streaming 处理成这样的格式 hadoop 1 spark 1 flink 1 kafka 1 hadoop 1 spark-streaming 1 6、使用mapWithState更新累积计数
该算子可以维护并更新每个key的状态。
这里用到一个新对象StateSpec且用到了它的两个方法initialState和function
initialState设置包含“mapWithState”将使用的初始状态的RDD
function设置实际的状态更新操作
//第1个参数状态 key 的类别
//第2个参数状态 value 的类别
//第3个参数状态 数据 的类别
//第4个参数状态 处理完要返回 的类别
def mappingFunction(key: String, value: Option[Int], state: State[Int]): Option[String] {// 使用state.exists()、state.get()、state.update()和state.remove()来管理状态并返回必要的字符串
}
四、运行
运行Netcat nc -lk 9999 新建一个窗口运行官方例子 cd /opt/cloudera/parcels/CDH-6.3.1-1.cdh6.3.1.p0.1470567/lib/spark/ bin/run-example org.apache.spark.examples.streaming.StatefulNetworkWordCount cdh1 9999 大多数高校硕博生毕业要求需要参加学术会议发表EI或者SCI检索的学术论文会议论文 可访问艾思科蓝官网浏览即将召开的学术会议列表。会议如下 第四届大数据、信息与计算机网络国际学术会议BDICN 2025
广州https://ais.cn/u/fi2yym
第四届电子信息工程、大数据与计算机技术国际学术会议EIBDCT 2025
青岛https://ais.cn/u/nuQr6f
第六届大数据与信息化教育国际学术会议ICBDIE 2025
苏州https://ais.cn/u/eYnmQr
第三届通信网络与机器学习国际学术会议(CNML 2025)
南京https://ais.cn/u/vUNva2
