Spark使用foreach与map算子的区别及自定义累加器的使用示例

如下代码，使用累加器对数据进行求和的过程中，发现代码没有报错，运行也无异常，但是累加器结果总是为空，经过多番查证，发现问题出现在下面代码的第六步。
开始时使用的是map算子对RDD中数据遍历，通过累加器进行计算，结果没有值。
后来发现map底层使用的是迭代器，循环遍历一遍之后，迭代器的指针已经跳到了最末尾，当进行累加器计算的时候，已经没有数据进行计算了，因此迭代器结果没有值。

修正后结果：使用foreach替代map进行循环计算。

spark使用map算子不产生数据的问题

package com.spark.exercise

import org.apache.spark.util.AccumulatorV2
import org.apache.spark.{SparkConf, SparkContext}

object $02_request1Update {
  
  def main(args: Array[String]): Unit = {
  
    //1.创建SparkContext
    val sc = new SparkContext(new SparkConf().setMaster("local[4]").setAppName("test"))
    //2.读取数据
    val datas = sc.textFile("datas/user_visit_action.txt")
    
    //3.创建累加器对象
    val acc = new SumAcc
    //4.注册累加器
    sc.register(acc)
  
    //5、切割数据,选择点击品类id，下单品类ids,支付品类ids,转换
    val source = datas.flatMap(line=>{
      val arr = line.split("_")
      //创建容器，封装数据
      var result:List[(String,Int,Int,Int)] = Nil
    
      val clickid = arr(6)
      if(clickid!="-1"){
        result = (clickid,1,0,0) :: result
      }else if( arr(8)!="null" ){
        arr(8).split(",").foreach(x=>{
          result = (x,0,1,0) :: result
        })
      }else if(arr(10)!="null"){
        arr(10).split(",").foreach(x=>{
          result = (x,0,0,1) :: result
        })
      }
      result
    } )
  
    //6.通过累加器累加数据
   source.foreach(x => acc.add(x))
  
   // println(acc.value.toBuffer)
  
    //7.获取累加结果，并排序取前十
    val result =  acc.value.toList.sortBy(_._2).reverse.take(10)
  
    println(result)
    
    Thread.sleep(100000)
  
  }
}

//定义累加器
class SumAcc extends AccumulatorV2[(String,Int,Int,Int),Map[String,(Int,Int,Int)]] {
  
  //创建一个容器，累加每个品类的点击次数、下单次数、支付次数
  var result = Map[String, (Int, Int, Int)]()
  
  //判断是否为空
  override def isZero: Boolean = result.isEmpty
  
  //赋值副本
  override def copy(): AccumulatorV2[(String, Int, Int, Int), Map[String, (Int, Int, Int)]] = new SumAcc
  
  //重置累加器
  override def reset(): Unit = Map[String, (Int, Int, Int)]()
  
  //累加元素
  override def add(v: (String, Int, Int, Int)): Unit = {
  
    //判断品类在容器中是否存在
    if(result.contains(v._1)){
    
      //取出之前的累加结果
      val value = result.getOrElse(v._1,(0,0,0))
      result = result.+((v._1,(value._1+v._2,value._2+v._3,value._3+v._4)))
    }else{
      result = result.+((v._1,(v._2,v._3,v._4)))
    }
  
  }
  
  //合并多个累加器的结果
  override def merge(other: AccumulatorV2[(String, Int, Int, Int), Map[String, (Int, Int, Int)]]): Unit = {
    //取出累加器的值
    val values = other.value
    
    values.foreach{
      case (id,(clicknum,ordernum,paynum)) =>
        //判断品类在容器中是否存在
        if(result.contains(id)){
      
          //取出之前的累加结果
          val value = result.getOrElse(id,(0,0,0))
          result = result.+((id,(value._1+clicknum,value._2+ordernum,value._3+paynum)))
        }else{
          result = result.+((id,(clicknum,ordernum,paynum)))
        }
    }
  }
  
  //返回值
  override def value: Map[String, (Int, Int, Int)] = this.result
  
}