spark广播变量的使用（转）

环境：
ubuntu16.04 64
伪分布式
使用的spark是2.3.1
scala 2.11.8
参考连接：
https://blog.csdn.net/android_xue/article/details/79780463#commentsedit

注意，这篇博客是对上述参考链接的总结和概括．

一句话讲明，广播变量干嘛的？
就是你代码里的某个变量在程序运行时太耗内存了，所以丢到各个slave机子中弄一个备份，省得代码运行时传来传去的，就是为了省一下内存以及传递的耗时．
下面上代码，
１.不用广播变量的完整代码BroadcastTest1.scala：

import scala.collection.Map
import scala.collection.mutable.ArrayBuffer
import scala.util.Random
import org.apache.spark.broadcast.Broadcast
import org.apache.spark.ml.recommendation.{ALS, ALSModel}
import org.apache.spark.sql.{DataFrame, Dataset, SparkSession}
import org.apache.spark.sql.functions._
import org.apache.log4j.Logger
import org.apache.log4j.Level
import org.apache.spark.sql.execution.datasources.textobject BroadcastTest
{def main(args: Array[String]): Unit = {Logger.getLogger("org").setLevel(Level.OFF)Logger.getLogger("akka").setLevel(Level.OFF)Logger.getRootLogger().setLevel(Level.ERROR) //这里是用来抑制一大堆log信息的.   val spark = SparkSession.builder.appName("Intro").config("spark.master", "local").getOrCreate();spark.sparkContext.setLogLevel("ERROR")val list=List("hello java lalala")val linesRDD= spark.read.textFile("hdfs://master:9000/test/word.txt")linesRDD.filter(line=>{list.contains(line)}).collect().foreach(println)}
}

运行方法:
1.启动Hadoop的HDFS系统
2.hdfs dfs -put word.txt hdfs://master:9000/test/
3.scalac BroadcastTest1.scala
4.scala BroadcastTest
当然，这里也可以采用maven打包用spark-submit来运行

2.用广播变量的完整代码BroadcastTest2.scala：

import scala.collection.Map
import scala.collection.mutable.ArrayBuffer
import scala.util.Random
import org.apache.spark.broadcast.Broadcast
import org.apache.spark.ml.recommendation.{ALS, ALSModel}
import org.apache.spark.sql.{DataFrame, Dataset, SparkSession}
import org.apache.spark.sql.functions._
import org.apache.log4j.Logger
import org.apache.log4j.Level
import org.apache.spark.sql.execution.datasources.textobject BroadcastTest
{def main(args: Array[String]): Unit = {Logger.getLogger("org").setLevel(Level.OFF)Logger.getLogger("akka").setLevel(Level.OFF)Logger.getRootLogger().setLevel(Level.ERROR) //这里是用来抑制一大堆log信息的.   val spark = SparkSession.builder.appName("Intro").config("spark.master", "local").getOrCreate();val sc = spark.sparkContextsc.setLogLevel("ERROR")val list=List("hello java lalala")val broadcast=sc.broadcast(list)val linesRDD= spark.read.textFile("hdfs://master:9000/test/word.txt")linesRDD.filter(line=> {broadcast.value.contains(line)}).collect().foreach(println)spark.stop()//这里也可以使用ｓｃ.stop()}}

运行方法:
1.启动Hadoop的HDFS系统
2.hdfs dfs -put word.txt hdfs://master:9000/test/
3.scalac BroadcastTest2.scala
4.scala BroadcastTest
当然，这里也可以采用maven打包用spark-submit来运行

spark广播变量的使用（转）相关推荐

Spark广播变量实现原理及基础编程
Spark广播变量实现原理及基础编程实现原理广播变量用来高效分发较大的对象.向所有工作节点发送一个较大的只读值,以供一个或多个 Spark 操作使用.比如,如果你的应用需要向所有节点发送一个较大的 ...
Spark广播变量使用示例
Spark广播变量使用示例实现原理广播变量用来高效分发较大的对象.向所有工作节点发送一个较大的只读值,以供一个或多个 Spark 操作使用.比如,如果你的应用需要向所有节点发送一个较大的只读查询表 ...
Spark广播变量之超大表left join小表时如何进行优化以及小表的正确位置
Spark广播变量之大表left join小表时如何进行优化以及小表的正确位置放置,带着这个目标我们一探究竟. 项目场景: 最近工作中遇到一个场景: 有一个超大表3.5T和一个小表963K 需要做关联 ...
spark广播变量的原理_spark使用广播变量
import java.io.{File, FileReader} import java.util import org.apache.spark.SparkConf import org.apac ...
spark 广播变量大数据_Spark基础知识（三）--- Spark的广播变量和累加器
在spark程序中,当一个传递给Spark操作(例如map和reduce)的函数在远程节点上面运行时,Spark操作实际上操作的是这个函数所用变量的一个独立副本.这些变量会被复制到每台机器上,并且这些 ...
spark广播变量和累加器
1 为什么使用广播变量和累加器变量存在的问题:在spark程序中,当一个传递给Spark操作(例如map和reduce)的函数在远程节点上面运行时,Spark操作实际上操作的是这个函数所用变量的 ...
spark学习-Spark广播变量与共享变量（1）
1,概念 ###1.1 广播变量: 广播变量允许程序员将一个只读的变量缓存在每台机器上,而不用在任务之间传递变量.广播变量可被用于有效地给每个节点一个大输入数据集的副本.Spark还尝试使用高效地广播 ...
Spark 广播变量和累加器
Spark 的一个核心功能是创建两种特殊类型的变量:广播变量和累加器广播变量(groadcast varible)为只读变量,它有运行SparkContext的驱动程序创建后发送给参与计算的节点.对 ...
Spark广播变量与累加器
在之前的文章中,我介绍了flink广播状态,从而了解了flink广播状态实际上就是将一个流广播到下游所有算子之中.在本文中我将介绍spark中类似的概念,为了方便理解,先放张spark应用程序架构图. ...

spark广播变量的使用（转）

spark广播变量的使用（转）相关推荐

最新文章

热门文章