Spark累加器、广播变量、案例需求

1. 累加器

累加器：分布式共享只写变量

考虑如下计算RDD中数据的和：

val rdd = sc.makeRDD(List(1, 2, 3, 4))

var sum = 0
rdd.foreach(
    num => {
        sum += num
    }
)

println("sum = " + sum)

预期结果10，但其实不是

foreach里面的函数是在Executor节点分布式执行的，所以多个分布式节点来同时修改sum，但是这个sum是Driver传给Executor的，各个Executor执行完并不会将sum返回给Driver，所以Driver端执行打印sum，sum依然为0。如果需要将最终的计算结果返回给Driver，就要使用到累加器变量。

累加器用来把Executor端变量信息聚合到Driver端。在Driver程序中定义的累加器变量，在Executor端会得到这个变量的一个新的副本，执行任务更新完这个副本之后，传回Driver端进行聚合。

使用累加器：

val rdd = sc.makeRDD(List(1, 2, 3, 4))

val sum = sc.longAccumulator("sum")

rdd.foreach(num => {
    sum.add(num)
})

println("sum = " + sum.value)

如果转换算子中使用了累加器，最终没有调用行动算子，那么实际也不会执行计算，例如下面代码sum依然为0

val rdd = sc.makeRDD(List(1, 2, 3, 4))

val sum = sc.longAccumulator("sum")

rdd.map(num => {
    sum.add(num)
    num
})

println("sum = " + sum.value)

注意，使用累加器最终执行行动算子不止一次，可能最终结果不符合预期

val rdd = sc.makeRDD(List(1, 2, 3, 4))

val sum = sc.longAccumulator("sum")

val mapRDD = rdd.map(num => {
    sum.add(num)
    num
})

mapRDD.collect()
mapRDD.collect()
println("sum = " + sum.value)

所以一般累加器需要放在行动算子里进行操作。

使用累加器实现WordCount，以避免Shuffle操作。首先定义一个自定义累加器，以实现map的累加（因为Spark中只有List集合的累加）

// AccumulatorV2泛型：IN：输入类型; OUT：输出类型
class MyAccumulator extends AccumulatorV2[String, mutable.Map[String, Long]] {
    private var wcMap = mutable.Map[String, Long]

    // 判断是否是初始状态
    override def isZero : Boolean = {
        wcMap.isEmpty
    }

    override def copy() : AccumulatorV2[String, mutable.Map[String, Long]] = {
        new MyAccumulator()
    }

    override def reset() : Unit = {
        wcMap.clear()
    }

    // 累加函数，每新来一个输入，如何累加
    override def add(word : String) : Unit = {
        val newCnt = wcMap.getOrElse(word, 0L) + 1
        wcMap.update(word, newCnt);
    }

    // 合并函数，由Driver执行，这里本质是两个map的合并
    override def merge(other : AccumulatorV2[String, mutable.Map[String, Long]]) : Unit = {
        val map1 = wcMap
        val map2 = other

        map2.foreach{
            case(word, count) => {
                val newCnt = map1.getOrElse(word, 0L) + count
                map1.update(word, newCnt)
            }
        }
    }

    override def value : mutable.Map[String, Long] = {
        wcMap
    }
}

val rdd = sc.makeRDD(List("hello", "spark", "hello"))

val myAcc = sc.newMyAccumulator()
sc.register(myAcc, "wordCountAcc")

rdd.foreach(word => {
    myAcc.add(word)
})

println(myAcc.value)

2. 广播变量

累加器：分布式共享只读变量

如果想实现两个rdd的相同key的value连接起来，一般会想到join算子，但是join是笛卡尔乘积，且存在shuffle，严重影响性能。可采用如下代码实现类似功能：

val rdd = sc.makeRDD(List(("a", 1), ("b", 2), ("c", 3)))

val map = mutable.Map(("a", 4), ("b", 5), ("c", 6))

rdd.map {
    case(w, c) => {
        val l : Int = map.getOrElse(w, 0)
        (w, (c, l))
    }
}.collect.foreach(println)

上述代码存在一个问题，map算子内的函数是在Executor内执行的（具体来说是Executor里的task执行的），里面用到了map这份数据，如果map很大，那么每个Executor的每个task都会包含这份大数据，非常占用内存，影响性能。于是引入了广播变量的概念，将这种数据存放在每一个Executor的内存中，该Executor中的各个task共享这个只读变量：

val rdd = sc.makeRDD(List(("a", 1), ("b", 2), ("c", 3)))

val map = mutable.Map(("a", 4), ("b", 5), ("c", 6))

val bc : Broadcast[mutable.Map[String, Int]] = sc.broadcast(map)

rdd1.map {
    case(w, c) => {
        val l : Int = bc.value.getOrElse(w, 0)
        (w, (c, l))
    }
}.collect.foreach(println)

3. 案例需求

1）已经从原始日志文件中读出了商品的点击数量rdd、下单数量rdd、支付数量rdd，都是(商品id, cnt)的形式，需要将这三种rdd组合成(商品id, (点击数量, 下单数量, 支付数量))的rdd，并且依次按照点击数量、下单数量、 支付数量排序取前十。

很自然地想到组合rdd的算子join，但是join只能组合相同的key，如果一个商品只有点击没有下单，那么使用join是不会出现在最终结果的，同理leftOuterJoin和rightOuterJoin也是类似的，不能实现相应的功能。因此只有cogroup算子满足要求，cogroup = connect + group。（join和cogroup算子的功能示例参见：RDD算子介绍（三））

val cogroupRDD : RDD[(String, (Iterable[Int], Iterable[Int], Iterable[Int]))]= clickRDD.cogroup(orderRDD, payRDD)

val resultRDD = cogroupRDD.mapValues{
    case(clickIter, orderIter, payIter) => {
        var clickCnt = 0
        val iter1 = clickIter.iterator
        if (iter1.hasNext) {
            clickCnt = iter1.next()
        }

        var orderCnt = 0
        val iter2 = orderIter.iterator
        if (iter2.hasNext) {
            orderCnt = iter2.next()
        }

        var payCnt = 0
        val iter3 = payIter.iterator
        if (iter3.hasNext) {
            payCnt = iter1.next()
        }

        (clickCnt, orderCnt, payCnt)
    }
}

resultRDD.sortBy(_._2, false).take(10).foreach(println)

上述实现方式使用了cogroup算子，该算子可能存在shuffle，且该算子不常用，可以采用另外一种方式实现。首先将商品的点击数量rdd、下单数量rdd、支付数量rdd转换为(商品id, (clickCnt, orderCnt, payCnt))的形式，然后进行union，最后进行聚合。

val rdd1 = clickRDD.map {
    case (id, cnt) => {
        (id, (cnt, 0, 0)
    }
}

val rdd2 = orderRDD.map {
    case (id, cnt) => {
        (id, (0, cnt, 0)
    }
}

val rdd3 = payRDD.map {
    case (id, cnt) => {
        (id, (0, 0, cnt)
    }
}

val rdd : RDD[(String, (Int, Int, Int))] = rdd1.union(rdd2).union(rdd3)

val resultRDD = rdd.reduceByKey((t1, t2) => {
    (t1._1 + t2._1, t1._2 + t2._2, t3._3 + t3._3)
})

resultRDD.sortBy(_._2, false).take(10).foreach(println)

上述代码使用了reduceByKey，还是会有shuffle操作，要完全避免shuffle操作，可以使用foreach算子，如果要使用foreach算子，就必须使用累加器了。这个累加器的作用就是遍历原始数据，按照商品id进行分组，对商品的用户行为类型（点击、下单、支付）对进行数量统计。可以将商品的用户行为（点击、下单、支付）数量封装成一个样例类HotCatagory，然后这个累加器的输入就是商品id+行为类型（点击、下单、支付），输出就是这个样例类的集合。具体实现过程参考https://www.bilibili.com/video/BV11A411L7CK?p=116&spm_id_from=pageDriver&vd_source=23ddeeb1fb342c5293413f7b87367160​​​​​​​

2）统计页面的跳转率。所谓某个页面A到某个页面B的跳转率，就是页面A到页面B的次数/页面A的点击次数。首先统计各个页面的点击次数，数据结构为map，作为分母。对于分子，需要按照sessionId进行分组，然后按照时间排序，这样才能得到各个用户浏览页面的顺序，然后转换数据结构，统计各个页面到其他页面的跳转次数。

actionDataRDD.cache()

// 分母
val pageIdToCountMap : Map[Long, Long] = actionDataRDD.map(action => {
    (action.page_id, 1L)
}).reduceByKey(_+_).collect.toMap

val sessionRDD : RDD[(String, Iterable[UserVisitAction])] = actionDataRDD.groupBy(_.session_id)

val mvRDD : RDD[(String, List[((Long, Long), Int)])] = sessionRDD.mapValues(
    iter => {
        val sortList : List[UserVisitAction] = iter.toList.sortBy(_.action_time)
        val flowIds : List[Long] = sortList.map(_.page_id)

        val pageFlowIds : List[(Long, Long)] = flowIds.zip(flowIds.tail)
        pageFlowIds.map(t=> (t, 1))
    }
)

// 分子
val dataRDD = mvRDD.map(_._2).flatMap(list=>list).reduceByKey(_+_)

dataRDD.foreach {
    case((page1, page2), sum) => {
        val cnt : Long = pageIdToCountMap.getOrElse(page1, 0L)
        println(s"页面${page1}跳转页面${page2}的转换率为" + (sum.toDouble / cnt))
    }
}