大数据-91 Spark 集群 RDD 编程-高阶 RDD广播变量 RDD累加器 Spark程序优化

点一下关注吧！！！非常感谢！！持续更新！！！

目前已经更新到了：

• Hadoop（已更完）
• HDFS（已更完）
• MapReduce（已更完）
• Hive（已更完）
• Flume（已更完）
• Sqoop（已更完）
• Zookeeper（已更完）
• HBase（已更完）
• Redis （已更完）
• Kafka（已更完）
• Spark（正在更新！）

章节内容

上节完成的内容如下：

• RDD容错机制
• RDD分区机制
• RDD分区器
• RDD自定义分区器

广播变量

基本介绍

有时候需要在多个任务之间共享变量，或者在任务（Task）和 Driver Program 之间共享变量。
为了满足这个需求，Spark提供了两种类型的变量。

• 广播变量（broadcast variable）
• 累加器（accumulators） 广播变量、累加器的主要作用是为了优化Spark程序。

广播变量将变量在节点的Executor之间进行共享（由Driver广播），广播变量用来高效分发较大的对象，向所有工作节点（Executor）发送一个较大的只读值，以供一个或多个操作使用。

使用广播变量的过程如下：

• 对一个类型T的对象调用SparkContext.broadcast创建一个Broadcast[T]对象，任何可序列化的类型都可以这么实现（在Driver端）
• 通过Value属性访问该对象的值（Executor中）
• 变量只会被分到各个Executor一次，作为只读值处理

广播变量的相关参数：

• spark.broadcast.blockSize（缺省值: 4m）
• spark.broadcast.checksum（缺省值：true）
• spark.broadcast.compree（缺省值：true）

变量应用

普通JOIN

MapSideJoin

生成数据 test_spark_01.txt

1000;商品1
1001;商品2
1002;商品3
1003;商品4
1004;商品5
1005;商品6
1006;商品7
1007;商品8
1008;商品9

生成数据格式如下：

生成数据 test_spark_02.txt

10000;订单1;100010001;订单2;100110002;订单3;100210003;订单4;100310004;订单5;100410005;订单6;100510006;订单7;100610007;订单8;100710008;订单9;1008

生成的数据格式如下：

编写代码1

我们编写代码进行测试

packageicu.wzkimportorg.apache.spark.rdd.RDD
importorg.apache.spark.{SparkConf, SparkContext}object JoinDemo {def main(args: Array[String]):Unit={val conf =new SparkConf().setAppName("JoinDemo").setMaster("local[*]")val sc =new SparkContext(conf)
    sc.hadoopConfiguration.setLong("fs.local.block.size",128*1024*1024)val productRDD: RDD[(String,String)]= sc
      .textFile("data/test_spark_01.txt").map {
        line =>val fields = line.split(";")(fields(0), line)}val orderRDD: RDD[(String,String)]= sc
      .textFile("data/test_spark_02.txt",8).map {
        line =>val fields = line.split(";")(fields(2), line)}val resultRDD = productRDD.join(orderRDD)
    println(resultRDD.count())
    Thread.sleep(100000)
    sc.stop()}}

编译打包1

mvn clean package

并上传到服务器，准备运行

运行测试1

spark-submit --master local[*]--class icu.wzk.JoinDemo spark-wordcount-1.0-SNAPSHOT.jar

提交任务并执行，注意数据的路径，查看下图：

运行结果可以查看到，运行了： 2.203100 秒 （取决于你的数据量的多少）

2024-07-19 10:35:08,808 INFO  [main] scheduler.DAGScheduler (Logging.scala:logInfo(54)) - Job 0 finished: count at JoinDemo.scala:32, took 2.203100 s
200

编写代码2

接下来，我们对比使用 MapSideJoin 的方式

packageicu.wzkimportorg.apache.spark.rdd.RDD
importorg.apache.spark.{SparkConf, SparkContext}object MapSideJoin {def main(args: Array[String]):Unit={val conf =new SparkConf().setAppName("MapSideJoin").setMaster("local[*]")val sc =new SparkContext(conf)
    sc.hadoopConfiguration.setLong("fs.local.block.size",128*1024*1024)val productRDD: RDD[(String,String)]= sc
      .textFile("data/test_spark_01.txt").map {
        line =>val fields = line.split(";")(fields(0), line)}val productBC = sc.broadcast(productRDD.collectAsMap())val orderRDD: RDD[(String,String)]= sc
      .textFile("data/test_spark_02.txt").map {
        line =>val fields = line.split(";")(fields(2), line)}val resultRDD = orderRDD
      .map {case(pid, orderInfo)=>val productInfo = productBC.value
          (pid,(orderInfo, productInfo.getOrElse(pid,null)))}
    println(resultRDD.count())

    sc.stop()}}

编译打包2

mvn clean package

编译后上传到服务器准备执行：

运行测试2

spark-submit --master local[*]--class icu.wzk.MapSideJoin spark-wordcount-1.0-SNAPSHOT.jar

启动我们的程序，并观察结果

我们可以观察到，这次只用了 0.10078 秒就完成了任务：

累加器

基本介绍

累加器的作用：可以实现一个变量在不同的Executor端能保持状态的累加。
累加器在Driver端定义、读取，在Executor中完成累加。
累加器也是Lazy的，需要Action触发：Action触发一次，执行一次；触发多次，执行多次。

Spark内置了三种类型的累加器，分别是：

• LongAccumulator 用来累加整数型
• DoubleAccumulator 用来累加浮点型
• CollectionAccumulator 用来累加集合元素

运行测试

我们可以在 SparkShell 中进行一些简单的测试，目前我在 h122 节点上，启动SparkShell

spark-shell --master local[*]

启动的主界面如下：

写入如下的内容进行测试：

val data = sc.makeRDD("hadoop spark hive hbase java scala hello world spark scala java hive".split("\\s+"))
val acc1 = sc.longAccumulator("totalNum1")
val acc2 = sc.doubleAccumulator("totalNum2")
val acc3 = sc.collectionAccumulator[String]("allwords")

我们进行测试的结果如下图所示：

继续编写一段进行测试：

val rdd = data.map{word => acc1.add(word.length); acc2.add(word.length); acc3.add(word); word}
rdd.count
rdd.collect

println(acc1.value)
println(acc2.value)
println(acc3.value)

我们进行测试的结果如下：