Spark 累加器

爱被打了一巴掌 2024-04-20 09:10 56阅读 0赞

# 5.RDD 编程进阶 #

## 5.1 累加器 ##

累加器用来对信息进行聚合，通常在向 Spark 传递函数时，比如使用 map() 函数或者

用 filter() 传条件时，可以使用驱动器程序中定义的变量，但是集群中运行的每个任务都会

得到这些变量的一份新的副本，更新这些副本的值也不会影响驱动器中的对应变量。如果我

们想实现所有分片处理时更新共享变量的功能，那么累加器可以实现我们想要的效果。

### 5.1.1 系统累加器 ###

针对一个输入的日志文件，如果我们想计算文件中所有空行的数量，我们可以编写以

下程序：

scala> val notice = sc.textFile("./NOTICE")
    notice: org.apache.spark.rdd.RDD[String] = ./NOTICE MapPartitionsRDD[40] at textFile at <console>:32
    
    scala> val blanklines = sc.accumulator(0)
    warning: there were two deprecation warnings; re-run with -deprecation for details
    blanklines: org.apache.spark.Accumulator[Int] = 0
    
    scala> val tmp = notice.flatMap(line => {
        | if (line == "") {
        | blanklines += 1
        | }
        | line.split(" ")
        | })
    tmp: org.apache.spark.rdd.RDD[String] = MapPartitionsRDD[41] at flatMap at <console>:36
    
    scala> tmp.count()
    res31: Long = 3213
    
    scala> blanklines.value
    res32: Int = 171

累加器的用法如下所示。

通过在驱动器中调用 SparkContext.accumulator(initialValue)方法，创建出存有初始值的

累加器。返回值为 org.apache.spark.Accumulator\[T\] 对象，其中 T 是初始值 initialValue 的

类型。Spark 闭包里的执行器代码可以使用累加器的 += 方法(在 Java 中是 add)增加累加器

的值。 驱动器程序可以调用累加器的 value 属性(在 Java 中使用 value()或 setValue())来访问

累加器的值。

注意： 工作节点上的任务不能访问累加器的值。从这些任务的角度来看，累加器是一个只写

变量。

对于要在行动操作中使用的累加器，Spark 只会把每个任务对各累加器的修改应用一次。

因此，如果想要一个无论在失败还是重复计算时都绝对可靠的累加器，我们必须把它放在

foreach() 这样的行动操作中。转化操作中累加器可能会发生不止一次更新

### 5.1.2 自定义累加器 ###

自定义累加器类型的功能在 1.X 版本中就已经提供了，但是使用起来比较麻烦，在 2.0

版本后，累加器的易用性有了较大的改进，而且官方还提供了一个新的抽象类：

AccumulatorV2 来提供更加友好的自定义类型累加器的实现方式。实现自定义类型累加器需

要继承 AccumulatorV2 并至少覆写下例中出现的方法，下面这个累加器可以用于在程序运行

过程中收集一些文本类信息，最终以 Set\[String\]的形式返回。

package com.lxl.spark
    
    import org.apache.spark.util.AccumulatorV2
    import org.apache.spark.{SparkConf, SparkContext}
    import scala.collection.JavaConversions._
    
    class LogAccumulator extends org.apache.spark.util.AccumulatorV2[String, java.util.Set[String]]
    {
      private val _logArray: java.util.Set[String] = new java.util.HashSet[String]()
      override def isZero: Boolean = {
        _logArray.isEmpty
      }
      override def reset(): Unit = {
        _logArray.clear()
      }
      override def add(v: String): Unit = {
        _logArray.add(v)
      }
      override def merge(other: org.apache.spark.util.AccumulatorV2[String, java.util.Set[String]]):
      Unit = {
        other match {
          case o: LogAccumulator => _logArray.addAll(o.value)
        }
      }
      override def value: java.util.Set[String] = {
        java.util.Collections.unmodifiableSet(_logArray)
      }
      override def copy():org.apache.spark.util.AccumulatorV2[String, java.util.Set[String]] = {
        val newAcc = new LogAccumulator()
        _logArray.synchronized{
          newAcc._logArray.addAll(_logArray)
        }
        newAcc
      }
    }
    // 过滤掉带字母的
    object LogAccumulator {
      def main(args: Array[String]) {
        val conf=new SparkConf().setAppName("LogAccumulator")
        val sc=new SparkContext(conf)
        val accum = new LogAccumulator
        sc.register(accum, "logAccum")
        val sum = sc.parallelize(Array("1", "2a", "3", "4b", "5", "6", "7cd", "8", "9"), 2).filter(line =>
        {
          val pattern = """^-?(\d+)"""
          val flag = line.matches(pattern)
          if (!flag) {
            accum.add(line)
          }
          flag
        }).map(_.toInt).reduce(_ + _)
        println("sum: " + sum)
        for (v <- accum.value) print(v + "")
        println()
        sc.stop()
      }
    }

笔记：

package com.lxl
    
    import java.util
    
    import org.apache.spark.util.AccumulatorV2
    import org.apache.spark.{SparkConf, SparkContext}
    
    /*
    自定义累加器
     */
    class CustomerAccu extends AccumulatorV2[String,java.util.Set[String]]{
    
      var logStr = new util.HashSet[String]()
    
      //判断当前对象是否为空
      override def isZero: Boolean = logStr.isEmpty
    
      //复制对象并返回
      override def copy(): AccumulatorV2[String, util.Set[String]] = {
    
        val accu = new CustomerAccu
        accu.logStr.addAll(logStr)
        accu
      }
    
      //重置对象
      override def reset(): Unit = logStr.clear()
    
      //分区内添加数据
      override def add(v: String): Unit = logStr.add(v)
    
      //分区间合合并
        override def merge(other: AccumulatorV2[String, util.Set[String]]): Unit = logStr.addAll(other.value)
    
      //返回值
      override def value: util.Set[String] = logStr
    }
    
    
    
    object Accu {
    
      def main(args: Array[String]): Unit = {
    
        //创建sparkConf对象
        val conf = new SparkConf().setAppName("Accu").setMaster("local[*]")
    
        //创建sparkcontext对象
        val sc = new SparkContext(conf)
        /*
      系统累加器
       */
        /*   var sum = sc.accumulator(0)
           val arr = Array(1,2,3,4,5)
           val rdd = sc.makeRDD(arr)
           rdd.map{ x =>
             sum += x
             x
           }.collect()
       //    for (i<-arr) sum += i
           println(sum.value) */
    
    
        /*
    自定义累加器使用
     */
        val rdd = sc.makeRDD(Array("a", "b", "c"))
    
        val accu = new CustomerAccu
        sc.register(accu, "CustomerAccu")
    
        rdd.map{ x =>
          accu.add(x)
          x
        }.collect()
    
        accu.value
        sc.stop()
    
      }
    }

## 5.2 广播变量（调优策略） ##

广播变量用来高效分发较大的对象。向所有工作节点发送一个较大的只读值，以供一

个或多个 Spark 操作使用。比如，如果你的应用需要向所有节点发送一个较大的只读查询

表，甚至是机器学习算法中的一个很大的特征向量，广播变量用起来都很顺手。 在多个并

行操作中使用同一个变量，但是 Spark 会为每个任务分别发送。

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

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

(1) 通过对一个类型 T 的对象调用 SparkContext.broadcast 创建出一个 Broadcast\[T\]对

象。任何可序列化的类型都可以这么实现。

(2) 通过 value 属性访问该对象的值(在 Java 中为 value()方法)。

(3) 变量只会被发到各个节点一次，应作为只读值处理(修改这个值不会影响到别的节

点)。

转载于:https://www.cnblogs.com/LXL616/p/11148328.html