标签：变量 instance sparkSession DataFrame RDD 广播
广播变量（BrocadCast）是Spark的一大特性，通过将小数据广播分发到每个执行任务的节点（Executor），从而避免了计算过程中的频繁拉去数据的网络带宽等开销。
Spark批处理和Spark Streaming流处理均支持广播变量。广播变量支持各种类型数据，包括数据、列表、Map、RDD、DataFrame等。
object BroadcastWrapper {
@volatile private var instance: Broadcast[Dataset[Row]] = null
def getHistoryData(sparkSession: SparkSession) : DataFrame = {
HiveUtil.initHiveEnv(sparkSession)
var maxPartition = HiveUtil.getTableMaxPartition(sparkSession, taiciTableName, "ds")
val historyDataDF = sparkSession.sql(s"select * from ${taiciTableName} where ds = ${maxPartition}")
historyDataDF
}
def updateBC(sc: SparkContext, sparkSession: SparkSession, blocking: Boolean = false): Unit = {
if (instance != null) {
instance.unpersist(blocking)
instance = sc.broadcast(getHistoryData(sparkSession).as("t_taici_history"))
}
}
//Broadcast[Dataset[Row]]
def getInstance(sc: SparkContext, sparkSession: SparkSession): Broadcast[Dataset[Row]] = {
if (instance == null) {
synchronized {
if (instance == null) {
instance = sc.broadcast(getHistoryData(sparkSession: SparkSession).as("t_taici_history"))
}
}
}
instance
}
}
流处理是时时刻刻在运行着的程序，很多时候广播变量需要定时更新读取一些数据，例如数据库配置等，这些数据会不定时更新，因此，广播变量也需要触发更新机制。在实践中，业务需要实时感知台词IP的死链变化和付费状态的变化，而从腾讯视频数据流得到的cid数据不一定在上线的台词IP中，因此需要实时判断新来的cid数据是否在已上线的台词中，而已经上线的台词数据是固定的，一两周才会更新一次，所以需要将已上线的台词数据作为广播变量分发到每个计算节点，这样计算的时候就避免了频繁的数据拉取，现在的全量台词数据量在70万，28MB，距离大的数据还有距离，故可以采取这种方式。
广播变量如果是DataFrame或者RDD，要注意，不能在RDD或者DataFrame的map或foreach算子中操作广播变量DataFrame。因为RDD或DataFrame本就是分布式执行的，因此在分布式节点是无法按照分布式算子操作广播变量RDD的，使用广播变量广播RDD的主要作用在于join关联的时候能够方便的关联，而避免了频繁的拉取数据。
标签：变量,instance,sparkSession,DataFrame,RDD,广播 来源： https://www.cnblogs.com/renyang/p/16487710.html
本站声明： 1. iCode9 技术分享网（下文简称本站）提供的所有内容，仅供技术学习、探讨和分享； 2. 关于本站的所有留言、评论、转载及引用，纯属内容发起人的个人观点，与本站观点和立场无关； 3. 关于本站的所有言论和文字，纯属内容发起人的个人观点，与本站观点和立场无关； 4. 本站文章均是网友提供，不完全保证技术分享内容的完整性、准确性、时效性、风险性和版权归属；如您发现该文章侵犯了您的权益，可联系我们第一时间进行删除； 5. 本站为非盈利性的个人网站，所有内容不会用来进行牟利，也不会利用任何形式的广告来间接获益，纯粹是为了广大技术爱好者提供技术内容和技术思想的分享性交流网站。