spark streaming中的广播变量应用

2024-06-16 19:58

本文主要是介绍spark streaming中的广播变量应用,希望对大家解决编程问题提供一定的参考价值,需要的开发者们随着小编来一起学习吧!

1. 广播变量

我们知道spark 的广播变量允许缓存一个只读的变量在每台机器上面,而不是每个任务保存一份拷贝。常见于spark在一些全局统计的场景中应用。通过广播变量,能够以一种更有效率的方式将一个大数据量输入集合的副本分配给每个节点。Spark也尝试着利用有效的广播算法去分配广播变量,以减少通信的成本。 
一个广播变量可以通过调用SparkContext.broadcast(v)方法从一个初始变量v中创建。广播变量是v的一个包装变量,它的值可以通过value方法访问,下面的代码说明了这个过程:

scala> val broadcastVar = sc.broadcast(Array(1, 2, 3))
broadcastVar: org.apache.spark.broadcast.Broadcast[Array[Int]] = Broadcast(0)scala> broadcastVar.value
res0: Array[Int] = Array(1, 2, 3)

2. Spark Streaming 广播变量的更新

广播变量的声明很简单,调用broadcast就能搞定,并且scala中一切可序列化的对象都是可以进行广播的,这就给了我们很大的想象空间,可以利用广播变量将一些经常访问的大变量进行广播,而不是每个任务保存一份,这样可以减少资源上的浪费。

但是,现在项目中遇到一种这样的需求,用spark streaming 通过一些离线全局更新好的数据对用户进行实时推荐(当然这里基于一些spark streaming的内部机制,不能实现真正的时效性):(1)日志流通过kafka获取 (2) 解析日志流数据,融合离线的全局数据,对每个Dtream进行计算(3)计算结果最后发送到redis中。

其中就会涉及这样的问题:(1)离线全局的数据是需要全局获取的,不能局部进行计算 (2)这部分数据是离线定期更新的,而spark streaming一旦开始,就长时间运行。如果离线数据更新了,如何在开始的流计算中,获取到这部分更新后的数据。

针对上述问题,我们可以直接想的一种方法是,在driver端开启一个附属线程,周期性去获取离线的全局数据,然后通过diver分发到各个task中。但是考虑到这种方式:spark streaming整体的性能开销会很大,并且重新开启的后台线程的不易管理。结合spark中的广播变量,我们采用另一种方式来解决以上问题: 
1> spark中的广播变量是只读的,通过unpersist函数,可以内存中的相关序列化对象 
2> 通过Dstream的foreachRDD方法,做到定时更新 (官网上有说明,该方法是在driver端执行的)


import java.io.{ObjectInputStream, ObjectOutputStream}
import com.bf.dt.wireless.config.WirelessConfig
import com.bf.dt.wireless.formator.WirelessFormator
import com.bf.dt.wireless.storage.MysqlConnectionPool
import com.bf.dt.wireless.utils.DateUtils
import kafka.serializer.StringDecoder
import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.broadcast.Broadcast
import org.apache.spark.streaming.kafka.KafkaUtils
import org.apache.spark.streaming.{Seconds, StreamingContext}
import org.json4s._
import org.slf4j.LoggerFactory
import scala.collection.mutableobject WirelessLogAnalysis {object BroadcastWrapper {@volatile private var instance: Broadcast[Map[String, List[String]]] = nullprivate val map = mutable.LinkedHashMap[String, List[String]]()def getMysql(): Map[String, List[String]] = {//1.获取mysql连接池的一个连接val conn = MysqlConnectionPool.getConnection.get//2.查询新的数据val sql = "select aid_type,aids from cf_similarity"val ps = conn.prepareStatement(sql)val rs = ps.executeQuery()while (rs.next()) {val aid = rs.getString("aid_type")val aids = rs.getString("aids").split(",").toListmap += (aid -> aids)}//3.连接池回收连接MysqlConnectionPool.closeConnection(conn)map.toMap}def update(sc: SparkContext, blocking: Boolean = false): Unit = {if (instance != null)instance.unpersist(blocking)instance = sc.broadcast(getMysql())}def getInstance(sc: SparkContext): Broadcast[Map[String, List[String]]] = {if (instance == null) {synchronized {if (instance == null) {instance = sc.broadcast(getMysql)}}}instance}private def writeObject(out: ObjectOutputStream): Unit = {out.writeObject(instance)}private def readObject(in: ObjectInputStream): Unit = {instance = in.readObject().asInstanceOf[Broadcast[Map[String, List[String]]]]}}def main(args: Array[String]): Unit = {val logger = LoggerFactory.getLogger(this.getClass)val conf = new SparkConf().setAppName("wirelessLogAnalysis")val ssc = new StreamingContext(conf, Seconds(10))val kafkaConfig: Map[String, String] = Map("metadata.broker.list" -> WirelessConfig.getConf.get.getString("wireless.metadata.broker.list"),"group.id" -> WirelessConfig.getConf.get.getString("wireless.group.id"),"zookeeper.connect" -> WirelessConfig.getConf.get.getString("wireless.zookeeper.connect"),"auto.offset.reset" -> WirelessConfig.getConf.get.getString("wireless.auto.offset.reset"))val androidvvTopic = WirelessConfig.getConf.get.getString("wireless.topic1")val iphonevvToplic = WirelessConfig.getConf.get.getString("wireless.topic2")val kafkaDStream = KafkaUtils.createDirectStream[String, String, StringDecoder, StringDecoder](ssc,kafkaConfig,Set(androidvvTopic, iphonevvToplic))//原始日志流打印kafkaDStream.print()val jsonDstream = kafkaDStream.map(x =>//解析日志流WirelessFormator.format(x._2))//解密的日志流打印jsonDstream.print()jsonDstream.foreachRDD {rdd => {// driver端运行,涉及操作:广播变量的初始化和更新// 可以自定义更新时间if ((DateUtils.getNowTime().split(" ")(1) >= "08:00:00") && (DateUtils.getNowTime().split(" ")(1) <= "10:10:00")) {BroadcastWrapper.update(rdd.sparkContext, true)println("广播变量更新成功: " + DateUtils.getNowTime())}//worker端运行,涉及操作:Dstream数据的处理和Redis更新rdd.foreachPartition {partitionRecords =>//1.获取redis连接,保证每个partition建立一次连接,避免每个记录建立/关闭连接的性能消耗partitionRecords.foreach(record => {//2.处理日志流val uid = record._1val aid_type = record._2 + "_" + record._3if (cf.value.keySet.contains(aid_type)) {(uid, cf.value.get(aid_type))println((uid, cf.value.get(aid_type)))}else(uid, "-1")}//3.redis更新数据)//4.关闭redis连接}}}ssc.start()ssc.awaitTermination()}
}

说明:以上是无线推荐项目中部分代码,其中离线全局数据存储在mysql中,MysqlConnectionPool是mysql连接池定义类,WirelessFormator是日志解密的定义类

这篇关于spark streaming中的广播变量应用的文章就介绍到这儿,希望我们推荐的文章对编程师们有所帮助!



http://www.chinasem.cn/article/1067403

相关文章

Python中随机休眠技术原理与应用详解

《Python中随机休眠技术原理与应用详解》在编程中,让程序暂停执行特定时间是常见需求,当需要引入不确定性时,随机休眠就成为关键技巧,下面我们就来看看Python中随机休眠技术的具体实现与应用吧... 目录引言一、实现原理与基础方法1.1 核心函数解析1.2 基础实现模板1.3 整数版实现二、典型应用场景2

Python Dash框架在数据可视化仪表板中的应用与实践记录

《PythonDash框架在数据可视化仪表板中的应用与实践记录》Python的PlotlyDash库提供了一种简便且强大的方式来构建和展示互动式数据仪表板,本篇文章将深入探讨如何使用Dash设计一... 目录python Dash框架在数据可视化仪表板中的应用与实践1. 什么是Plotly Dash?1.1

Android Kotlin 高阶函数详解及其在协程中的应用小结

《AndroidKotlin高阶函数详解及其在协程中的应用小结》高阶函数是Kotlin中的一个重要特性,它能够将函数作为一等公民(First-ClassCitizen),使得代码更加简洁、灵活和可... 目录1. 引言2. 什么是高阶函数?3. 高阶函数的基础用法3.1 传递函数作为参数3.2 Lambda

Java中&和&&以及|和||的区别、应用场景和代码示例

《Java中&和&&以及|和||的区别、应用场景和代码示例》:本文主要介绍Java中的逻辑运算符&、&&、|和||的区别,包括它们在布尔和整数类型上的应用,文中通过代码介绍的非常详细,需要的朋友可... 目录前言1. & 和 &&代码示例2. | 和 ||代码示例3. 为什么要使用 & 和 | 而不是总是使

Python循环缓冲区的应用详解

《Python循环缓冲区的应用详解》循环缓冲区是一个线性缓冲区,逻辑上被视为一个循环的结构,本文主要为大家介绍了Python中循环缓冲区的相关应用,有兴趣的小伙伴可以了解一下... 目录什么是循环缓冲区循环缓冲区的结构python中的循环缓冲区实现运行循环缓冲区循环缓冲区的优势应用案例Python中的实现库

SpringBoot整合MybatisPlus的基本应用指南

《SpringBoot整合MybatisPlus的基本应用指南》MyBatis-Plus,简称MP,是一个MyBatis的增强工具,在MyBatis的基础上只做增强不做改变,下面小编就来和大家介绍一下... 目录一、MyBATisPlus简介二、SpringBoot整合MybatisPlus1、创建数据库和

python中time模块的常用方法及应用详解

《python中time模块的常用方法及应用详解》在Python开发中,时间处理是绕不开的刚需场景,从性能计时到定时任务,从日志记录到数据同步,时间模块始终是开发者最得力的工具之一,本文将通过真实案例... 目录一、时间基石:time.time()典型场景:程序性能分析进阶技巧:结合上下文管理器实现自动计时

Java逻辑运算符之&&、|| 与&、 |的区别及应用

《Java逻辑运算符之&&、||与&、|的区别及应用》:本文主要介绍Java逻辑运算符之&&、||与&、|的区别及应用的相关资料,分别是&&、||与&、|,并探讨了它们在不同应用场景中... 目录前言一、基本概念与运算符介绍二、短路与与非短路与:&& 与 & 的区别1. &&:短路与(AND)2. &:非短

Spring AI集成DeepSeek三步搞定Java智能应用的详细过程

《SpringAI集成DeepSeek三步搞定Java智能应用的详细过程》本文介绍了如何使用SpringAI集成DeepSeek,一个国内顶尖的多模态大模型,SpringAI提供了一套统一的接口,简... 目录DeepSeek 介绍Spring AI 是什么?Spring AI 的主要功能包括1、环境准备2

Spring AI与DeepSeek实战一之快速打造智能对话应用

《SpringAI与DeepSeek实战一之快速打造智能对话应用》本文详细介绍了如何通过SpringAI框架集成DeepSeek大模型,实现普通对话和流式对话功能,步骤包括申请API-KEY、项目搭... 目录一、概述二、申请DeepSeek的API-KEY三、项目搭建3.1. 开发环境要求3.2. mav