标签：转换 val Streaming org apache import spark Spark ssc
1、有状态转换
1.UpdateStateByKey
updateStateByKey 操作使得我们可以在用新信息进行更新时保持任意的状态。为使用这个功能，你需要做下面两步：
-
定义状态，状态可以是一个任意的数据类型。
-
定义状态更新函数，用此函数阐明如何使用之前的状态和来自输入流的新值对状态进行更新。
使用 updateStateByKey 需要对检查点目录进行配置，会使用检查点来保存状态。
2.编写代码
package com.zch.spark.streaming
import org.apache.kafka.clients.consumer.{ConsumerConfig, ConsumerRecord}
import org.apache.spark.SparkConf
import org.apache.spark.streaming.dstream.{InputDStream, ReceiverInputDStream}
import org.apache.spark.streaming.kafka010.{ConsumerStrategies, KafkaUtils, LocationStrategies}
import org.apache.spark.streaming.{Seconds, StreamingContext}
/**
* Author: zhaoHui
* Date: 2022/01/03
* Time: 19:03
* Description: 有状态操作
*/
object sparkStreaming05_State {
def main(args: Array[String]): Unit = {
val sparkConf = new SparkConf()
.setMaster("local[*]")
.setAppName("state")
val ssc = new StreamingContext(sparkConf, Seconds(3))
ssc.checkpoint("F:\\JAVA\\bigdata2107\\zch\\spark\\src\\main\\resources\\sparkStreamingStates")
// 无状态操作，只对当前的采集周期内的数据进行处理
// 在某些场合下，需要保留数据统计结果（状态），实现数据汇总
// 使用有状态操作时，需要设置checkpoint路径
val datas: ReceiverInputDStream[String] = ssc.socketTextStream("localhost", 9999)
val wordToOne = datas.map((_, 1))
// updateStateByKey：根据key对数据的状态进行更新
// 传递的参数中含有两个值
// 第一个值表示相同的key的value数据
// 第二个值表示缓存区相同key的value数据
val state = wordToOne.updateStateByKey(
(seq:Seq[Int],buff:Option[Int]) => {
val newCount = buff.getOrElse(0) + seq.sum
Option(newCount)
}
)
state.print()
ssc.start()
ssc.awaitTermination()
}
}2、优雅的关闭
优雅的关闭：计算节点不在接收新的数据，而是将现有的数据处理完毕后，然后关闭。
package com.zch.spark.streaming
import org.apache.spark.SparkConf
import org.apache.spark.streaming.dstream.{DStream, ReceiverInputDStream}
import org.apache.spark.streaming.{Seconds, StreamingContext, StreamingContextState}
/**
* Author: zhaoHui
* Date: 2022/01/03
* Time: 19:03
* Description: 有状态操作
*/
object sparkStreaming05_State_Close {
def main(args: Array[String]): Unit = {
val sparkConf = new SparkConf()
.setMaster("local[*]")
.setAppName("close")
val ssc = new StreamingContext(sparkConf, Seconds(3))
val lines: ReceiverInputDStream[String] = ssc.socketTextStream("localhost", 9999)
val wordToOne: DStream[(String, Int)] = lines.map((_, 1))
wordToOne.print()
ssc.start()
// 如果想要关闭采集器，就要启动一个新的线程
// 而且需要在第三方增加关闭状态
new Thread(
() => {
// 优雅的关闭
// 计算节点不在接收新的数据，而是将现有的数据处理完毕后，然后关闭
// Mysql
// Redis
// 等等
while (true){
if (true){ // 相当于mysql存入数据了
// 获取SparkStreaming状态
val state: StreamingContextState = ssc.getState()
if (state == StreamingContextState.ACTIVE){
ssc.stop(stopSparkContext = true, stopGracefully = true)
}
}
Thread.sleep(5000)
System.exit(0)
}
}
).start()
ssc.awaitTermination()
}
}标签：转换,val,Streaming,org,apache,import,spark,Spark,ssc 来源： https://blog.csdn.net/chaohui2638457321/article/details/122362155
本站声明： 1. iCode9 技术分享网（下文简称本站）提供的所有内容，仅供技术学习、探讨和分享； 2. 关于本站的所有留言、评论、转载及引用，纯属内容发起人的个人观点，与本站观点和立场无关； 3. 关于本站的所有言论和文字，纯属内容发起人的个人观点，与本站观点和立场无关； 4. 本站文章均是网友提供，不完全保证技术分享内容的完整性、准确性、时效性、风险性和版权归属；如您发现该文章侵犯了您的权益，可联系我们第一时间进行删除； 5. 本站为非盈利性的个人网站，所有内容不会用来进行牟利，也不会利用任何形式的广告来间接获益，纯粹是为了广大技术爱好者提供技术内容和技术思想的分享性交流网站。