Spark广播变量、累加器、WebUI

朱雀 2021-11-26 14:26 310阅读 0赞

# Spark广播变量、累加器、WebUI #

## 一、广播变量 ##

### 使用原因及优点 ###

Driver提交N个task到Excutor，Excutor会产生N个存放task的内存，换句话说就是，发送多少个task，就携带多少个Driver端副本，为了解决这种问题，需要用到广播变量

当给Driver端的变量贴上broadcast标签后，Excutor会产生一个BlockManager来管理broadcast，待到下一个task任务提交给Excutor时，task会先向BlockManager寻找是否存在相关的broadcast，如果有就直接使用。

优点：Driver提交N个task到Excutor，Excutor会产生一个内存来处理

### 原理图 ###

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQwOTI5OTIx_size_16_color_FFFFFF_t_70]

### 注意事项 ###

广播变量只能在Driver定义初始化，在Executor使用，不能在Excutor中改变广播变量的值

不能将RDD广播出去，可以将结果广播出去，rdd.collect()

### 代码 ###

val conf = new SparkConf().setAppName("broadCastTest").setMaster("local")
    val sc = new SparkContext(conf)
    
    //添加广播变量
    val list = List[String]("zhangsan","lisi")
    val bcList: Broadcast[List[String]] = sc.broadcast(list)
    val nameRDD = sc.parallelize(List[String]("zhangsan","lisi","wangwu"))
    
    val result = nameRDD.filter(name=>{
      val innerList:List[String] = bcList.value
      !innerList.contains(name)
    })
    
    result.foreach(println)

## 二、累加器 ##

### 使用原因 ###

为了统计各个Executor的执行数量，使用累加器后各个Excutor上执行的数据数，可以返回Driver端，并进行相关计算

不使用累加器的情况下，各个Executor数据数无法汇总到Driver端，无法显示

### 原理图 ###

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQwOTI5OTIx_size_16_color_FFFFFF_t_70 1]

### 注意事项 ###

累加器在Driver端定义初始化

### 代码 ###

val conf : SparkSession = SparkSession.builder().appName("accumunlatorTest").master("local").getOrCreate()
    val sc : SparkContext = conf.sparkContext
    sc.setLogLevel("Error")
    val rdd1: RDD[String] = sc.textFile("./Test_spark/data/words2")
    
    val accumulator: LongAccumulator = sc.longAccumulator
    val rdd2: RDD[String] = rdd1.map(one => {
      accumulator.add(1)
      one
    })
    rdd2.collect()
    println(s"accumulator = ${accumulator.value}" )

## 三、WebUI ##

### 操作 ###

打开hdfs集群

在Master节点的spark/sbin下 ./start-all.sh

在上传节点的spark/bin下 ./spark-shell --master spark://sxt001:7077

运行相关代码(以WC为例) sc.textFile(“hdfs://sxt001:8020/spark/data/words.txt”).flatMap(*.split(" ")).map((*,1)).reduceByKey(*+*).collect()

### 查看1-Application ID ###

1 打开8888端口查看

\[外链图片转存失败(img-ra5JJ2m5-1563934387694)(C:\\Users\\86158\\AppData\\Roaming\\Typora\\typora-user-images\\1563783441876.png)\]

2 点开application是job的信息界面

\[外链图片转存失败(img-PqAASVva-1563934387694)(C:\\Users\\86158\\AppData\\Roaming\\Typora\\typora-user-images\\1563783572830.png)\]

3 继续下个页面是Excutor信息界面

\[外链图片转存失败(img-ve176aVU-1563934387695)(C:\\Users\\86158\\AppData\\Roaming\\Typora\\typora-user-images\\1563783626805.png)\]

### 查看2-Name ###

\[外链图片转存失败(img-Ub1erOWs-1563934387695)(C:\\Users\\86158\\AppData\\Roaming\\Typora\\typora-user-images\\1563783441876.png)\]

点击name查看当前任务的详情，每个导航栏导航栏都有详细信息

\[外链图片转存失败(img-ZSnJ2Wbk-1563934387695)(C:\\Users\\86158\\AppData\\Roaming\\Typora\\typora-user-images\\1563783974802.png)\]

## 四、WebUI历史数据服务器配置 ##

在spark/conf cp文件 vim spark-defaults.conf

添加 spark.eventLog.enabled true  
 spark.eventLog.dir hdfs://bjsxt/spark/log  
 spark.history.fs.logDirectory hdfs://bjsxt/spark/log

spark.eventLog.compress true

启动spark/sbin ./start-history-server.sh

查看历史日志 打开传输节点：18080端口查看

xt/spark/log

spark.eventLog.compress true

启动spark/sbin ./start-history-server.sh

查看历史日志 打开传输节点：18080端口查看

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQwOTI5OTIx_size_16_color_FFFFFF_t_70]: /images/20211126/9b803c5bac77420d8ee383f0d1181cdf.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQwOTI5OTIx_size_16_color_FFFFFF_t_70 1]: /images/20211126/9e4ca8b316584cdcac2482dff7380a02.png