标签搜索

目 录CONTENT

文章目录

Spark核心编程(Scala语言)

陈铭
2023-01-30 / 0 评论 / 0 点赞 / 167 阅读 / 10,075 字 / 正在检测是否收录...

RDD

Spark 计算框架为了能够进行高并发和高吞吐的数据处理,封装了三大数据结构,用于处理不同的应用场景。三大数据结构分别是:

  • RDD : 弹性分布式数据集
  • 累加器:分布式共享只写变量
  • 广播变量:分布式共享只读变量

什么是RDD

RDD(Resilient Distributed Dataset)叫做弹性分布式数据集,是 Spark 中最基本的数据处理模型。代码中是一个抽象类,它代表一个弹性的、不可变、可分区、里面的元素可并行计算的集合。类似于Hadoop编程需要写Mapper、Reducer和Driver,Spark内计算任务的封装就是RDD。

RDD的运行流程

在Spark架构中,任何一个计算任务,都会被RDD进行拆分粒度更细的Task,放在一个TaskPool中。通过Yarn的调度,获取合适的NameNode,NN内启动容器,容器内的Executor会获取Task进行并行计算。
image
image

RDD核心编程

由于Spark是由Scala语言(在JVM上运行,可以视作Java增强版)编写的,对于计算任务的编程此处也必须使用Scala。

WordCount的样例

这里抛砖引玉一个大数据经典案例

package ltd.cmjava.spark.wordcount

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

object WordCount {
  def main(args: Array[String]): Unit = {
    // 创建 Spark 运行配置对象
    val sparkConf = new SparkConf().setMaster("local[*]").setAppName("WordCount")
    // 创建 Spark 上下文环境对象(连接对象)
    val sc: SparkContext = new SparkContext(sparkConf)
    // textFile从文件读取RDD
    // 一行一行从文件读取数据
    // [["hello scala"], ["hello spark"]],(文件里就这两行数据)
    // sc.textFile("hdfs://xxxx:xxx/xxx/xxx/xxx.txt"),也可以从HDFS读取
    val fileRDD: RDD[String] = sc.textFile("D:\\Develop\\workspace\\Spark\\spark-core\\src\\main\\resources\\wordcount\\word.txt")
    // 对空格进行切割,然后扁平化
    // [["hello", "scala"], ["hello", "spark"]]=>["hello", "hello", "scala", "spark"]
    val wordRDD: RDD[String] = fileRDD.flatMap(_.split(" "))
    // 转换数据结构 [("hello", 1), ("hello", 1), ("scala", 1), ("spark", 1)]
    val word2OneRDD: RDD[(String, Int)] = wordRDD.map((_, 1))
    // 将转换结构后的数据按照相同的单词进行分组聚合
    // [("hello", 1), ("hello", 1), ("scala", 1), ("spark", 1)]
    // =>[("hello", 2), ("scala", 1), ("spark", 1)]
    val word2CountRDD: RDD[(String, Int)] = word2OneRDD.reduceByKey(_ + _)
    // 将数据聚合结果收集转换为Array
    val word2Count: Array[(String, Int)] = word2CountRDD.collect()
    // 打印结果
    word2Count.foreach(println)
    // 关闭 Spark 连接
    sc.stop()
  }
}

RDD的创建

从内存读取数据创建RDD

package ltd.cmjava.spark.core.rdd.builder

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

object Spark01_RDD_Memory {

    def main(args: Array[String]): Unit = {

        // local[*]使用本地环境的所有核
        // local[2]使用本地环境的2核
        val sparkConf = new SparkConf().setMaster("local[*]").setAppName("RDD")
        val sc = new SparkContext(sparkConf)

        // 创建RDD
        // 从内存中创建RDD,将内存中集合的数据作为处理的数据源
        val seq = Seq[Int](1,2,3,4)

        // parallelize : 并行
        // val rdd: RDD[Int] = sc.parallelize(seq)
        // makeRDD方法在底层实现时其实就是调用了rdd对象的parallelize方法。
        val rdd: RDD[Int] = sc.makeRDD(seq)

        rdd.collect().foreach(println)

        // 关闭环境
        sc.stop()
    }
}

从文件读取数据创建RDD

以textFile方式读取 : 以行为单位来读取数据,读取的数据都是字符串

package ltd.cmjava.spark.core.rdd.builder

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

object Spark02_RDD_File {

    def main(args: Array[String]): Unit = {
        // 准备环境
        val sparkConf = new SparkConf().setMaster("local[*]").setAppName("RDD")
        val sc = new SparkContext(sparkConf)

        // 从文件中创建RDD,将文件中的数据作为处理的数据源
        // path路径默认以当前环境的根路径为基准。可以写绝对路径,也可以写相对路径
        //sc.textFile("D:\\mineworkspace\\idea\\classes\\atguigu-classes\\datas\\1.txt")
        //val rdd: RDD[String] = sc.textFile("datas/1.txt")

        // path路径可以是文件的具体路径,也可以目录名称
        //val rdd = sc.textFile("datas")

        // path路径还可以使用通配符 *
        //val rdd = sc.textFile("datas/1*.txt")

        // path还可以是分布式存储系统路径:HDFS
        val rdd = sc.textFile("hdfs://hadoop100:8020/test.txt")
        rdd.collect().foreach(println)

        // 关闭环境
        sc.stop()
    }
}

以wholeTextFiles方式读取 : 以文件为单位读取数据

package ltd.cmjava.spark.core.rdd.builder

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

object Spark02_RDD_File1 {

    def main(args: Array[String]): Unit = {

        // 准备环境
        val sparkConf = new SparkConf().setMaster("local[*]").setAppName("RDD")
        val sc = new SparkContext(sparkConf)

        // 创建RDD
        // 从文件中创建RDD,将文件中的数据作为处理的数据源

        // textFile : 以行为单位来读取数据,读取的数据都是字符串
        // wholeTextFiles : 以文件为单位读取数据
        //    读取的结果表示为元组,第一个元素表示文件路径,第二个元素表示文件内容
        val rdd = sc.wholeTextFiles("datas")

        rdd.collect().foreach(println)

        // 关闭环境
        sc.stop()
    }
}

RDD的分区

从内存读取数据创建RDD

package ltd.cmjava.spark.core.rdd.builder


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

import java.io.File
import scala.reflect.io.Directory

object Spark01_RDD_Memory_Par {

    def main(args: Array[String]): Unit = {

        // TODO 准备环境
        val sparkConf = new SparkConf().setMaster("local[*]").setAppName("RDD")
        sparkConf.set("spark.default.parallelism", "5")
        val sc = new SparkContext(sparkConf)

        // TODO 创建RDD
        // RDD的并行度 & 分区
        // makeRDD方法可以传递第二个参数,这个参数表示分区的数量
        // 第二个参数可以不传递的,那么makeRDD方法会使用默认值 : defaultParallelism(默认并行度)
        //    defaultParallelism = scheduler.conf.getInt("spark.default.parallelism", totalCores)
        //    spark在默认情况下,从配置对象中获取配置参数:spark.default.parallelism
        //    如果获取不到,那么使用totalCores属性,这个属性取值为当前运行环境的最大可用核数
        val rdd = sc.makeRDD(List(1,2,3,4))
        
        // 【1,2】,【3,4】
        // val rdd = sc.makeRDD(List(1,2,3,4), 2)
        // 【1】,【2】,【3,4】
        // val rdd = sc.makeRDD(List(1,2,3,4), 3)
        // 【1】,【2,3】,【4,5】
        // val rdd = sc.makeRDD(List(1,2,3,4,5), 3)

        // 将处理的数据保存成分区文件
        new Directory(new File("output")).deleteRecursively()
        rdd.saveAsTextFile("output")

        // TODO 关闭环境
        sc.stop()
    }
}

从文件读取数据创建RDD

package ltd.cmjava.spark.core.rdd.builder

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

import java.io.File
import scala.reflect.io.Directory

object Spark02_RDD_File_Par {

    def main(args: Array[String]): Unit = {

        // 准备环境
        val sparkConf = new SparkConf().setMaster("local[*]").setAppName("RDD")
        val sc = new SparkContext(sparkConf)

        // 创建RDD
        // textFile可以将文件作为数据处理的数据源,默认也可以设定分区。
        //     minPartitions : 最小分区数量
        //     minPartitions = math.min(defaultParallelism, 2)
        //     defaultParallelism = 当前运行环境的最大可用核数
        // 如果不想使用默认的分区数量,可以通过第二个参数指定分区数
        // Spark读取文件,底层其实使用的就是Hadoop的读取方式

        // 分区数量的计算方式(datas/1.txt为7byte的文件):
        //    totalSize = 7 byte (totalSize,总共的文件大小)
        //    goalSize =  7 / 2 = 3 byte(goalSize,每个分区分配的大小;2,分区数)
        //    7 / 3 = 2余1 (因此最后输出3个分区)
        new Directory(new File("output")).deleteRecursively()
        val rdd = sc.textFile("datas/1.txt", 2)

        rdd.saveAsTextFile("output")


        // 关闭环境
        sc.stop()
    }
}

RDD的转换算子

RDD 根据数据处理方式的不同将算子整体上分为Value 类型、双 Value 类型和Key-Value类型。转换算子并不会触发Job的运行,只有行动算子才行!!

map算子

函数签名

def map[U: ClassTag](f: T => U): RDD[U]

函数说明
将处理的数据逐条进行映射转换,这里的转换可以是类型的转换,也可以是值的转换。但map不会对原来数据的数量进行改变

    val dataRDD: RDD[Int] = sparkContext.makeRDD(List(1, 2, 3, 4))
    val dataRDD1: RDD[Int] = dataRDD.map(
      num => {
        num * 2
      }
    )
    val dataRDD2: RDD[String] = dataRDD1.map(
      num => {
        "" + num
      }
    )

mapPartitions算子

函数签名

def mapPartitions[U: ClassTag]( f: Iterator[T] => Iterator[U],
preservesPartitioning: Boolean = false): RDD[U]

函数说明
将待处理的数据以分区为单位发送到计算节点进行处理,这里的处理是指可以进行任意的处理,哪怕是过滤数据。

    val dataRDD1: RDD[Int] = dataRDD.mapPartitions(
      datas => {
        datas.filter(_ == 2)
      }
    )

map 和 mapPartitions 的区别?

  1. 数据处理角度
    Map 算子是分区内一个数据一个数据的执行,类似于串行操作。而 mapPartitions 算子是以分区为单位进行批处理操作。
  2. 功能的角度
    Map 算子主要目的将数据源中的数据进行转换和改变。但是不会减少或增多数据。 MapPartitions 算子需要传递一个迭代器,返回一个迭代器,没有要求的元素的个数保持不变,所以可以增加或减少数据
  3. 性能的角度
    Map 算子因为类似于串行操作,所以性能比较低,而是 mapPartitions 算子类似于批处
    理,所以性能较高。但是mapPartitions 算子会长时间占用内存,那么这样会导致内存可能不够用,出现内存溢出的错误。所以在内存有限的情况下,不推荐使用。使用 map 操作。

mapPartitionsWithIndex算子

函数签名

def mapPartitionsWithIndex[U: ClassTag]( f: (Int, Iterator[T]) => Iterator[U],
preservesPartitioning: Boolean = false): RDD[U]

函数说明
将待处理的数据以分区为单位发送到计算节点进行处理,这里的处理是指可以进行任意的处理,哪怕是过滤数据,在处理时同时可以获取当前分区索引。

    val dataRDD1 = dataRDD.mapPartitionsWithIndex(
      (index, datas) => {
        datas.map(index, _)
      }
    )

flatMap算子

函数签名

def flatMap[U: ClassTag](f: T => TraversableOnce[U]): RDD[U]

函数说明
将处理的数据进行扁平化后再进行映射处理,所以算子也称之为扁平映射

    val dataRDD = sparkContext.makeRDD(
      List(List(1, 2), List(3, 4)), 1)
    val dataRDD1 = dataRDD.flatMap(
      list => list
    )

glom 算子

函数签名

def glom(): RDD[Array[T]]

函数说明
将同一个分区的数据直接转换为相同类型的内存数组进行处理,分区不变

    val dataRDD = sparkContext.makeRDD(List(1, 2, 3, 4), 1)
    val dataRDD1: RDD[Array[Int]] = dataRDD.glom()

groupBy算子

函数签名

def groupBy[K](f: T => K)(implicit kt: ClassTag[K]): RDD[(K, Iterable[T])]

函数说明
将数据根据指定的规则进行分组, 分区默认不变,但是数据会被打乱重新组合,我们将这样的操作称之为shuffle。极限情况下,数据可能被分在同一个分区中
**注意:groupBy算子不会改变分区数量,但是数据会被分组。一个完整的组的数据在一个分区中,但是并不是说一个分区中只有一个组。
**

    val dataRDD = sparkContext.makeRDD(List(1, 2, 3, 4), 1)
    val dataRDD1 = dataRDD.groupBy(
      _ % 2
    )

filter算子

函数签名

def filter(f: T => Boolean): RDD[T]

函数说明
将数据根据指定的规则进行筛选过滤,符合规则的数据保留,不符合规则的数据丢弃。
当数据进行筛选过滤后,分区不变,只是分区内不符合条件的数据被抛弃了。但是分区内的数据可能不均衡,生产环境下,可能会出现数据倾斜。

    val dataRDD = sparkContext.makeRDD(List(1, 2, 3, 4), 1)
    val dataRDD1 = dataRDD.filter(_ % 2 == 0)

sample算子

函数签名

def sample( withReplacement: Boolean, fraction: Double, seed: Long = Utils.random.nextLong): RDD[T]

函数说明
根据指定的规则从数据集中抽取数据

    val dataRDD = sparkContext.makeRDD(List(1, 2, 3, 4), 1)
    // 抽取数据不放回(伯努利算法)
    // 伯努利算法:又叫 0、1 分布。例如扔硬币,要么正面,要么反面。
    // 具体实现:根据种子和随机算法算出一个数和第二个参数设置几率比较,小于第二个参数要,大于不要
    // 第一个参数:抽取的数据是否放回,false:不放回
    // 第二个参数:抽取的几率,范围在[0,1]之间,0:全不取;1:全取;
    // 第三个参数:随机数种子
    val dataRDD1 = dataRDD.sample(false, 0.5)
    // 抽取数据放回(泊松算法)
    // 第一个参数:抽取的数据是否放回,true:放回;false:不放回
    // 第二个参数:重复数据的几率,范围大于等于 0.表示每一个元素被期望抽取到的次数
    // 第三个参数:随机数种子
    val dataRDD2 = dataRDD.sample(true, 2)

distinct算子

函数签名

def distinct()(implicit ord: Ordering[T] = null): RDD[T]
def distinct(numPartitions: Int)(implicit ord: Ordering[T] = null): RDD[T]

函数说明
将数据集中重复的数据去重

    val dataRDD = sparkContext.makeRDD(List( 1,2,3,4,1,2),1)
    val dataRDD1 = dataRDD.distinct()
    val dataRDD2 = dataRDD.distinct(2)

coalesce算子

函数签名

def coalesce(numPartitions: Int, shuffle: Boolean = false, partitionCoalescer: Option[PartitionCoalescer] = Option.empty) (implicit ord: Ordering[T] = null): RDD[T]

函数说明
根据数据量缩减分区,用于大数据集过滤后,提高小数据集的执行效率
当 spark 程序中,存在过多的小任务的时候,可以通过 coalesce 方法,收缩合并分区,减少分区的个数,减小任务调度成本

    val dataRDD = sparkContext.makeRDD(List( 1,2,3,4,1,2),6)
    val dataRDD1 = dataRDD.coalesce(2)

repartition算子

函数签名

def repartition(numPartitions: Int)(implicit ord: Ordering[T] = null): RDD[T]

函数说明
该操作内部其实执行的是 coalesce 操作,参数 shuffle 的默认值为 true。无论是将分区数多的 RDD 转换为分区数少的RDD,还是将分区数少的 RDD 转换为分区数多的RDD,repartition操作都可以完成,因为无论如何都会经 shuffle 过程。

    val dataRDD = sparkContext.makeRDD(List( 1,2,3,4,1,2),2)
    val dataRDD1 = dataRDD.repartition(4)

sortBy算子

函数签名

def sortBy[K]( f: (T) => K, ascending: Boolean = true, numPartitions: Int = this.partitions.length)(implicit ord: Ordering[K], ctag: ClassTag[K]): RDD[T]

函数说明
该操作用于排序数据。在排序之前,可以将数据通过 f 函数进行处理,之后按照 f 函数处理的结果进行排序,默认为升序排列。排序后新产生的 RDD 的分区数与原RDD 的分区数一致。中间存在 shuffle 的过程

    val dataRDD = sparkContext.makeRDD(List( 1,2,3,4,1,2),2)
    val dataRDD1 = dataRDD.sortBy(num=>num, false, 4)

intersection算子

函数签名

def intersection(other: RDD[T]): RDD[T]

函数说明
对源RDD 和参数RDD 求交集后返回一个新的RDD

    val dataRDD1 = sparkContext.makeRDD(List(1,2,3,4)) 
    val dataRDD2 = sparkContext.makeRDD(List(3,4,5,6)) 
    val dataRDD = dataRDD1.intersection(dataRDD2)

union算子

函数签名

def union(other: RDD[T]): RDD[T]

函数说明
对源RDD 和参数RDD 求并集后返回一个新的RDD

    val dataRDD1 = sparkContext.makeRDD(List(1,2,3,4)) 
    val dataRDD2 = sparkContext.makeRDD(List(3,4,5,6))
    val dataRDD = dataRDD1.union(dataRDD2)

subtract算子

函数签名

def subtract(other: RDD[T]): RDD[T]

函数说明
以一个 RDD 元素为主,去除两个 RDD 中重复元素,将其他元素保留下来。求差集

    val dataRDD1 = sparkContext.makeRDD(List(1,2,3,4)) 
    val dataRDD2 = sparkContext.makeRDD(List(3,4,5,6)) 
    val dataRDD = dataRDD1.subtract(dataRDD2)

zip算子

函数签名

def zip[U: ClassTag](other: RDD[U]): RDD[(T, U)]

函数说明
将两个 RDD 中的元素,以键值对的形式进行合并。其中,键值对中的Key 为第 1 个 RDD中的元素,Value 为第 2 个 RDD 中的相同位置的元素。

    val dataRDD1 = sparkContext.makeRDD(List(1,2,3,4)) 
    val dataRDD2 = sparkContext.makeRDD(List(3,4,5,6)) 
    val dataRDD = dataRDD1.zip(dataRDD2)

partitionBy算子

函数签名

def partitionBy(partitioner: Partitioner): RDD[(K, V)]

函数说明
将数据按照指定Partitioner 重新进行分区。Spark 默认的分区器是HashPartitioner

    val rdd: RDD[(Int, String)] = sc.makeRDD(Array((1,"aaa"),(2,"bbb"),(3,"ccc")),3)
    import org.apache.spark.HashPartitioner
    val rdd2: RDD[(Int, String)] = rdd.partitionBy(new HashPartitioner(2))

reduceByKey算子

函数签名

def reduceByKey(func: (V, V) => V): RDD[(K, V)]
def reduceByKey(func: (V, V) => V, numPartitions: Int): RDD[(K, V)]

函数说明
可以将数据按照相同的Key 对Value 进行聚合

    val dataRDD1 = sparkContext.makeRDD(List(("a",1),("b",2),("c",3))) 
    val dataRDD2 = dataRDD1.reduceByKey(_+_)
    val dataRDD3 = dataRDD1.reduceByKey(_+_, 2)

groupByKey算子

函数签名

def groupByKey(): RDD[(K, Iterable[V])]
def groupByKey(numPartitions: Int): RDD[(K, Iterable[V])]
def groupByKey(partitioner: Partitioner): RDD[(K, Iterable[V])]

函数说明
将数据源的数据根据 key 对 value 进行分组

    val dataRDD1 = sparkContext.makeRDD(List(("a",1),("b",2),("c",3)))
    val dataRDD2 = dataRDD1.groupByKey() 
    val dataRDD3 = dataRDD1.groupByKey(2)
    val dataRDD4 = dataRDD1.groupByKey(new HashPartitioner(2))

reduceByKey 和 groupByKey 的区别?
1. 从 shuffle 的角度
reduceByKey 和 groupByKey 都存在 shuffle 的操作,但是reduceByKey可以在 shuffle 前对分区内相同 key 的数据进行预聚合(combine)功能,这样会减少落盘的数据量,而groupByKey 只是进行分组,不存在数据量减少的问题,reduceByKey 性能比较高。
2. 从功能的角度
reduceByKey 其实包含分组和聚合的功能。GroupByKey 只能分组,不能聚合,所以在分组聚合的场合下,推荐使用 reduceByKey,如果仅仅是分组而不需要聚合。那么还是只能使用groupByKey

aggregateByKey算子

函数签名

def aggregateByKey[U: ClassTag](zeroValue: U)(seqOp: (U, V) => U, combOp: (U, U) => U): RDD[(K, U)]

函数说明
将数据根据不同的规则进行分区内计算和分区间计算

    // TODO : 取出每个分区内相同 key 的最大值然后分区间相加
    // aggregateByKey 算子是函数柯里化,存在两个参数列表
    // 1. 第一个参数列表中的参数表示初始值
    // 2. 第二个参数列表中含有两个参数
    //	2.1 第一个参数表示分区内的计算规则
    //	2.2 第二个参数表示分区间的计算规则
    val rdd =
    sc.makeRDD(List( ("a",1),("a",2),("c",3),
      ("b",4),("c",5),("c",6)
    ),2)
    // 0:("a",1),("a",2),("c",3) => (a,10)(c,10)
    //	=> (a,10)(b,10)(c,20)
    // 1:("b",4),("c",5),("c",6) => (b,10)(c,10)
    val resultRDD = rdd.aggregateByKey(10)(
      (x, y) => math.max(x,y), (x, y) => x + y
    )
    resultRDD.collect().foreach(println)

foldByKey算子

函数签名

def foldByKey(zeroValue: V)(func: (V, V) => V): RDD[(K, V)]

函数说明
当分区内计算规则和分区间计算规则相同时,aggregateByKey 就可以简化为foldByKey

    val dataRDD1 = sparkContext.makeRDD(List(("a",1),("b",2),("c",3))) 
    val dataRDD2 = dataRDD1.foldByKey(0)(_+_)

combineByKey算子

函数签名

def combineByKey[C]( createCombiner: V => C, mergeValue: (C, V) => C, mergeCombiners: (C, C) => C): RDD[(K, C)]

函数说明
最通用的对key-value 型 rdd 进行聚集操作的聚集函数(aggregation function)。类似于 aggregate(),combineByKey()允许用户返回值的类型与输入不一致。

    val list: List[(String, Int)] = List(("a", 88), ("b", 95), ("a", 91), ("b", 93),
      ("a", 95), ("b", 98))
    val input: RDD[(String, Int)] = sc.makeRDD(list, 2)

    val combineRdd: RDD[(String, (Int, Int))] = input.combineByKey( (_, 1),
      (acc: (Int, Int), v) => (acc._1 + v, acc._2 + 1),
      (acc1: (Int, Int), acc2: (Int, Int)) => (acc1._1 + acc2._1, acc1._2 + acc2._2)
    )

reduceByKey、foldByKey、aggregateByKey、combineByKey 的区别?
reduceByKey: 相同 key 的第一个数据不进行任何计算,分区内和分区间计算规则相同
FoldByKey: 相同 key 的第一个数据和初始值进行分区内计算,分区内和分区间计算规则相同
AggregateByKey: 相同 key 的第一个数据和初始值进行分区内计算,分区内和分区间计算规则可以不相同
CombineByKey: 当计算时,发现数据结构不满足要求时,可以让第一个数据转换结构。分区内和分区间计算规则不相同。

sortByKey算子

函数签名

def sortByKey(ascending: Boolean = true, numPartitions: Int = self.partitions.length): RDD[(K, V)]

函数说明
在一个(K,V)的 RDD 上调用,K 必须实现 Ordered 接口(特质),返回一个按照 key 进行排序的

    val dataRDD1 = sparkContext.makeRDD(List(("a",1),("b",2),("c",3))) 
    val sortRDD1: RDD[(String, Int)] = dataRDD1.sortByKey(true)
    val sortRDD1: RDD[(String, Int)] = dataRDD1.sortByKey(false)

join算子

函数签名

  def join[W](other: RDD[(K, W)]): RDD[(K, (V, W))]

函数说明
在类型为(K,V)和(K,W)的RDD 上调用,返回一个相同 key 对应的所有元素连接在一起的(K,(V,W))的RDD

    val rdd: RDD[(Int, String)] = sc.makeRDD(Array((1, "a"), (2, "b"), (3, "c")))
    val rdd1: RDD[(Int, Int)] = sc.makeRDD(Array((1, 4), (2, 5), (3, 6))) 
    rdd.join(rdd1).collect().foreach(println)

leftOuterJoin算子

函数签名

  def leftOuterJoin[W](other: RDD[(K, W)]): RDD[(K, (V, Option[W]))]

函数说明
类似于 SQL 语句的左外连接

    val dataRDD1 = sparkContext.makeRDD(List(("a",1),("b",2),("c",3)))
    val dataRDD2 = sparkContext.makeRDD(List(("a",1),("b",2),("c",3)))
    val rdd: RDD[(String, (Int, Option[Int]))] = dataRDD1.leftOuterJoin(dataRDD2)

cogroup算子

函数签名

def cogroup[W](other: RDD[(K, W)]): RDD[(K, (Iterable[V], Iterable[W]))]

函数说明
在类型为(K,V)和(K,W)的RDD 上调用,返回一个(K,(Iterable<V>,Iterable<W>))类型的 RDD

    val dataRDD1 = sparkContext.makeRDD(List(("a",1),("a",2),("c",3)))
    val dataRDD2 = sparkContext.makeRDD(List(("a",1),("c",2),("c",3))) 
    val value: RDD[(String, (Iterable[Int], Iterable[Int]))] = dataRDD1.cogroup(dataRDD2)

RDD的行动算子

reduce算子

函数签名

def reduce(f: (T, T) => T): T

函数说明
聚集RDD 中的所有元素,先聚合分区内数据,再聚合分区间数据

    val rdd: RDD[Int] = sc.makeRDD(List(1,2,3,4))
    // 聚合数据
    val reduceResult: Int = rdd.reduce(_+_)

collect算子

函数签名

def collect(): Array[T]

函数说明
在驱动程序中,以数组Array 的形式返回数据集的所有元素

    val rdd: RDD[Int] = sc.makeRDD(List(1,2,3,4))
    // 收集数据到 Driver 
    rdd.collect().foreach(println)

count算子

函数签名

def count(): Long

函数说明
返回RDD 中元素的个数

    val rdd: RDD[Int] = sc.makeRDD(List(1,2,3,4))
    // 返回 RDD 中元素的个数
    val countResult: Long = rdd.count()

first算子

函数签名

def first(): T

函数说明
返回RDD 中的第一个元素

    val rdd: RDD[Int] = sc.makeRDD(List(1,2,3,4))
    // 返回 RDD 中元素的个数
    val firstResult: Int = rdd.first()

take算子

函数签名

def take(num: Int): Array[T]

函数说明
返回一个由RDD 的前 n 个元素组成的数组

    val rdd: RDD[Int] = sc.makeRDD(List(1,2,3,4))
    // 返回 RDD 中元素的个数
    val takeResult: Array[Int] = rdd.take(2)

takeOrdered算子

函数签名

def takeOrdered(num: Int)(implicit ord: Ordering[T]): Array[T]

函数说明
返回该 RDD 排序后的前 n 个元素组成的数组

    val rdd: RDD[Int] = sc.makeRDD(List(1,3,2,4))
    // 返回 RDD 中元素的个数
    val result: Array[Int] = rdd.takeOrdered(2)

aggregate算子

函数签名

def aggregate[U: ClassTag](zeroValue: U)(seqOp: (U, T) => U, combOp: (U, U) => U): U

函数说明
分区的数据通过初始值和分区内的数据进行聚合,然后再和初始值进行分区间的数据聚合

    val rdd: RDD[Int] = sc.makeRDD(List(1, 2, 3, 4), 8)
    // 将该 RDD 所有元素相加得到结果
    //val result: Int = rdd.aggregate(0)(_ + _, _ + _)
    val result: Int = rdd.aggregate(10)(_ + _, _ + _)

fold算子

函数签名

def fold(zeroValue: T)(op: (T, T) => T): T

函数说明
折叠操作,aggregate 的简化版操作

    val rdd: RDD[Int] = sc.makeRDD(List(1, 2, 3, 4))
    val foldResult: Int = rdd.fold(0)(_+_)

countByKey算子

函数签名

def countByKey(): Map[K, Long]

函数说明
统计每种 key 的个数

    val rdd: RDD[(Int, String)] = sc.makeRDD(List((1, "a"), (1, "a"), (1, "a"), (2, "b"), (3, "c"), (3, "c")))
    // 统计每种 key 的个数
    val result: collection.Map[Int, Long] = rdd.countByKey()

save 相关算子

函数签名

def saveAsTextFile(path: String): Unit def saveAsObjectFile(path: String): Unit def saveAsSequenceFile(path: String, codec: Option[Class[_ <: CompressionCodec]] = None): Unit

函数说明
将数据保存到不同格式的文件中

    // 保存成 Text 文件
    rdd.saveAsTextFile("output")
    // 序列化成对象保存到文件
    rdd.saveAsObjectFile("output1")
    // 保存成 Sequencefile 文件
    rdd.map((_,1)).saveAsSequenceFile("output2")

foreach算子

函数签名

def foreach(f: T => Unit): Unit = withScope { val cleanF = sc.clean(f)
    sc.runJob(this, (iter: Iterator[T]) => iter.foreach(cleanF))
}

函数说明
分布式遍历RDD 中的每一个元素,调用指定函数

    val rdd: RDD[Int] = sc.makeRDD(List(1,2,3,4))
    // 收集后打印
    rdd.map(num=>num).collect().foreach(println) println("****************")
    // 分布式打印
    rdd.foreach(println)

RDD的序列化

闭包检查

从计算的角度, 算子以外的代码都是在Driver 端执行, 算子里面的代码都是在 Executor端执行。那么在 scala 的函数式编程中,就会导致算子内经常会用到算子外的数据,这样就形成了闭包的效果,如果使用的算子外的数据无法序列化,就意味着无法传值给Executor端执行,就会发生错误,所以需要在执行任务计算前,检测闭包内的对象是否可以进行序列化,这个操作我们称之为闭包检测。Scala2.12 版本后闭包编译方式发生了改变

序列化方法和属性

从计算的角度, 算子以外的代码都是在Driver 端执行, 算子里面的代码都是在 Executor端执行,看如下代码:

package ltd.cmjava.spark.core.rdd.serial

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

object Spark01_RDD_Serial {

    def main(args: Array[String]): Unit = {
        val sparConf = new SparkConf().setMaster("local").setAppName("WordCount")
        val sc = new SparkContext(sparConf)

        val rdd: RDD[String] = sc.makeRDD(Array("hello world", "hello spark", "hive", "atguigu"))

        val search = new Search("h")

//        search.getMatch1(rdd).collect().foreach(println)
        println("======================")
        search.getMatch2(rdd).collect().foreach(println)

        sc.stop()
    }
    // 查询对象
    // 类的构造参数其实是类的属性, 构造参数需要进行闭包检测,其实就等同于类进行闭包检测
    class Search(query:String){

        def isMatch(s: String): Boolean = {
            s.contains(this.query)
        }

        // 函数序列化案例
        def getMatch1 (rdd: RDD[String]): RDD[String] = {
            rdd.filter(isMatch)
        }

        // 属性序列化案例
        def getMatch2(rdd: RDD[String]): RDD[String] = {
            val s = query
            rdd.filter(x => x.contains(s))
        }
    }
}

Kryo 序列化框架

Java 的序列化能够序列化任何的类。但是比较重(字节多),序列化后,对象的提交也比较大。Spark 出于性能的考虑,Spark2.0 开始支持另外一种Kryo 序列化机制。Kryo 速度是 Serializable 的 10 倍。当 RDD 在 Shuffle 数据的时候,简单数据类型、数组和字符串类型已经在 Spark 内部使用 Kryo 来序列化。
注意: 即使使用Kryo 序列化,也要继承Serializable 接口。

package ltd.cmjava.spark.core.rdd.serial

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

object serializable_Kryo {
  // 替换默认的序列化机制
  // 注册需要使用 kryo 序列化的自定义类
  // conf 替换默认的序列化机制
  def main(args: Array[String]): Unit = {
    val conf: SparkConf = new SparkConf()
      .setAppName("SerDemo")
      .setMaster("local[*]")
      .set("spark.serializer", "org.apache.spark.serializer.KryoSerializer") // 注册需要使用 kryo 序列化的自定义类
      .registerKryoClasses(Array(classOf[Searcher]))
    val sc = new SparkContext(conf)
    val rdd: RDD[String] = sc.makeRDD(Array("hello world", "hello atguigu", "atguigu", "hahah"), 2)

    val searcher = new Searcher("hello")
    val result: RDD[String] = searcher.getMatchedRDD1(rdd)

    result.collect.foreach(println)
  }

  // 注意:即使使用Kryo 序列化,也要继承Serializable 接口
  class Searcher (val query: String) extends Serializable{

    def isMatch(s: String) = {
      s.contains(query)
    }

    def getMatchedRDD1(rdd: RDD[String]) = {
      rdd.filter(isMatch)
    }

    def getMatchedRDD2(rdd: RDD[String]) = {
      val q = query
      rdd.filter(_.contains(q))
    }
  }
}

RDD的依赖关系

RDD的血缘关系

RDD 只支持粗粒度转换,即在大量记录上执行的单个操作。将创建 RDD 的一系列Lineage(血统)记录下来,以便恢复丢失的分区。RDD 的Lineage 会记录RDD 的元数据信息和转换行为,当该RDD 的部分分区数据丢失时,它可以根据这些信息来重新运算和恢复丢失的数据分区。

package ltd.cmjava.spark.core.rdd.dep

import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.rdd.RDD

object Spark01_RDD_Dep {

    def main(args: Array[String]): Unit = {

        val sparConf = new SparkConf().setMaster("local").setAppName("WordCount")
        val sc = new SparkContext(sparConf)

        val lines: RDD[String] = sc.textFile("datas/word.txt")
        println(lines.toDebugString)
        println("*************************")
        val words: RDD[String] = lines.flatMap(_.split(" "))
        println(words.toDebugString)
        println("*************************")
        val wordToOne = words.map(word=>(word,1))
        println(wordToOne.toDebugString)
        println("*************************")
        val wordToSum: RDD[(String, Int)] = wordToOne.reduceByKey(_+_)
        println(wordToSum.toDebugString)
        println("*************************")
        val array: Array[(String, Int)] = wordToSum.collect()
        array.foreach(println)

        sc.stop()

    }
}

RDD的依赖关系

这里所谓的依赖关系,其实就是两个相邻 RDD 之间的关系

package ltd.cmjava.spark.core.rdd.dep

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

object Spark02_RDD_Dep {

    def main(args: Array[String]): Unit = {

        val sparConf = new SparkConf().setMaster("local").setAppName("Dep")
        val sc = new SparkContext(sparConf)

        val lines: RDD[String] = sc.textFile("datas/word.txt")
        println(lines.dependencies)
        println("*************************")
        val words: RDD[String] = lines.flatMap(_.split(" "))
        println(words.dependencies)
        println("*************************")
        val wordToOne = words.map(word=>(word,1))
        println(wordToOne.dependencies)
        println("*************************")
        val wordToSum: RDD[(String, Int)] = wordToOne.reduceByKey(_+_)
        println(wordToSum.dependencies)
        println("*************************")
        val array: Array[(String, Int)] = wordToSum.collect()
        array.foreach(println)

        sc.stop()

    }
}

RDD的窄依赖和宽依赖

窄依赖表示每一个父(上游)RDD 的 Partition 最多被子(下游)RDD 的一个 Partition 使用,窄依赖我们形象的比喻为独生子女。

class OneToOneDependency[T](rdd: RDD[T]) extends NarrowDependency[T](rdd)	

宽依赖表示同一个父(上游)RDD 的 Partition 被多个子(下游)RDD 的 Partition 依赖,会引起 Shuffle,总结:宽依赖我们形象的比喻为多生。

class ShuffleDependency[K: ClassTag, V: ClassTag, C: ClassTag]( 	
	@transient private val _rdd: RDD[_ <: Product2[K, V]],
	val partitioner: Partitioner,
	val serializer: Serializer = SparkEnv.get.serializer, 
    val keyOrdering: Option[Ordering[K]] = None,
	val aggregator: Option[Aggregator[K, V, C]] = None, 
    val mapSideCombine: Boolean = false)
	extends Dependency[Product2[K, V]]

RD的任务划分

RDD 任务切分中间分为:Application、Job、Stage 和 Task
Application: 初始化一个 SparkContext 即生成一个Application;
Job: 一个Action 算子就会生成一个Job;
Stage: Stage 等于宽依赖(ShuffleDependency)的个数加 1;
Task: 一个 Stage 阶段中,最后一个RDD 的分区个数就是Task 的个数。
**注意:**Application->Job->Stage->Task 每一层都是 1 对 n 的关系。

RDD的持久化

RDD的Cache缓存

RDD 通过Cache 或者 Persist 方法将前面的计算结果缓存,默认情况下会把数据以缓存在 JVM 的堆内存中。但是并不是这两个方法被调用时立即缓存,而是触发后面的 action 算子时,该RDD 将会被缓存在计算节点的内存中,并供后面重用。

package ltd.cmjava.spark.core.rdd.persist

import org.apache.spark.rdd.RDD
import org.apache.spark.storage.StorageLevel
import org.apache.spark.{SparkConf, SparkContext}

object Spark03_RDD_Persist {

    def main(args: Array[String]): Unit = {
        val sparConf = new SparkConf().setMaster("local").setAppName("Persist")
        val sc = new SparkContext(sparConf)

        val list = List("Hello Scala", "Hello Spark")

        val rdd = sc.makeRDD(list)

        val flatRDD = rdd.flatMap(_.split(" "))

        val mapRDD = flatRDD.map(word=>{
            println("@@@@@@@@@@@@")
            (word,1)
        })
        // cache默认持久化的操作,只能将数据保存到内存中,如果想要保存到磁盘文件,需要更改存储级别
        //mapRDD.cache()

        // 持久化操作必须在行动算子执行时完成的。
        mapRDD.persist(StorageLevel.DISK_ONLY)

        val reduceRDD: RDD[(String, Int)] = mapRDD.reduceByKey(_+_)
        reduceRDD.collect().foreach(println)
        println("**************************************")
        val groupRDD = mapRDD.groupByKey()
        groupRDD.collect().foreach(println)


        sc.stop()
    }
}

存储级别

object StorageLevel {
    val NONE = new StorageLevel(false, false, false, false) val DISK_ONLY = new StorageLevel(true, false, false, false)
    val DISK_ONLY_2 = new StorageLevel(true, false, false, false, 2) val MEMORY_ONLY = new StorageLevel(false, true, false, true)
    val MEMORY_ONLY_2 = new StorageLevel(false, true, false, true, 2) val MEMORY_ONLY_SER = new StorageLevel(false, true, false, false)
    val MEMORY_ONLY_SER_2 = new StorageLevel(false, true, false, false, 2) val MEMORY_AND_DISK = new StorageLevel(true, true, false, true)
    val MEMORY_AND_DISK_2 = new StorageLevel(true, true, false, true, 2) val MEMORY_AND_DISK_SER = new StorageLevel(true, true, false, false)
    val MEMORY_AND_DISK_SER_2 = new StorageLevel(true, true, false, false, 2) val OFF_HEAP = new StorageLevel(true, true, true, false, 1)
}

image

缓存有可能丢失,或者存储于内存的数据由于内存不足而被删除,RDD 的缓存容错机制保证了即使缓存丢失也能保证计算的正确执行。通过基于RDD 的一系列转换,丢失的数据会被重算,由于RDD 的各个 Partition 是相对独立的,因此只需要计算丢失的部分即可,并不需要重算全部Partition。
Spark 会自动对一些 Shuffle 操作的中间数据做持久化操作(比如:reduceByKey)。这样做的目的是为了当一个节点 Shuffle 失败了避免重新计算整个输入。但是,在实际使用的时候,如果想重用数据,仍然建议调用 persist 或 cache。

RDD的CheckPoint检查点

所谓的检查点其实就是通过将RDD 中间结果写入磁盘
由于血缘依赖过长会造成容错成本过高,这样就不如在中间阶段做检查点容错,如果检查点之后有节点出现问题,可以从检查点开始重做血缘,减少了开销。
对 RDD 进行 checkpoint 操作并不会马上被执行,必须执行行动算子的操作才能触发。

package ltd.cmjava.spark.core.rdd.persist

import org.apache.spark.rdd.RDD
import org.apache.spark.storage.StorageLevel
import org.apache.spark.{SparkConf, SparkContext}

object Spark04_RDD_Persist {

    def main(args: Array[String]): Unit = {
        val sparConf = new SparkConf().setMaster("local").setAppName("Persist")
        val sc = new SparkContext(sparConf)
        // 可以写hdfs的路径:hdfs://ip:port/xx/xx/xx
        sc.setCheckpointDir("cp")

        val list = List("Hello Scala", "Hello Spark")

        val rdd = sc.makeRDD(list)

        val flatRDD = rdd.flatMap(_.split(" "))

        val mapRDD = flatRDD.map(word=>{
            println("@@@@@@@@@@@@")
            (word,1)
        })
        // checkpoint 需要落盘,需要指定检查点保存路径
        // 检查点路径保存的文件,当作业执行完毕后,不会被删除
        // 一般保存路径都是在分布式存储系统:HDFS
        mapRDD.checkpoint()

        val reduceRDD: RDD[(String, Int)] = mapRDD.reduceByKey(_+_)
        reduceRDD.collect().foreach(println)
        println("**************************************")
        val groupRDD = mapRDD.groupByKey()
        groupRDD.collect().foreach(println)


        sc.stop()
    }
}

缓存和检查点区别

  1. Cache 缓存只是将数据保存起来,不切断血缘依赖。Checkpoint 检查点切断血缘依赖。

  2. Cache 缓存的数据通常存储在磁盘、内存等地方,可靠性低。Checkpoint 的数据通常存储在HDFS 等容错、高可用的文件系统,可靠性高。

  3. 建议对checkpoint()的RDD 使用Cache 缓存,这样 checkpoint 的 job 只需从 Cache 缓存中读取数据即可,否则需要再从头计算一次RDD。

RDD的分区器

Spark 目前支持Hash 分区和 Range 分区,和用户自定义分区。Hash 分区为当前的默认分区。分区器直接决定了RDD 中分区的个数、RDD 中每条数据经过Shuffle 后进入哪个分区,进而决定了Reduce 的个数。
只有Key-Value 类型的RDD 才有分区器,非 Key-Value 类型的RDD 分区的值是 None
每个RDD 的分区 ID 范围:0 ~ (numPartitions - 1),决定这个值是属于那个分区的。

Hash分区器

对于给定的 key,计算其hashCode,并除以分区个数取余

Range分区器

将一定范围内的数据映射到一个分区中,尽量保证每个分区数据均匀,而且分区间有序

自定义分区器

package ltd.cmjava.spark.core.rdd.part

import org.apache.spark.rdd.RDD
import org.apache.spark.{HashPartitioner, Partitioner, SparkConf, SparkContext}

object Spark01_RDD_Part {

    def main(args: Array[String]): Unit = {
        val sparConf = new SparkConf().setMaster("local").setAppName("WordCount")
        val sc = new SparkContext(sparConf)

        val rdd = sc.makeRDD(List(
            ("nba", "xxxxxxxxx"),
            ("cba", "xxxxxxxxx"),
            ("wnba", "xxxxxxxxx"),
            ("nba", "xxxxxxxxx"),
        ),3)
        val partRDD: RDD[(String, String)] = rdd.partitionBy( new MyPartitioner() )

        partRDD.saveAsTextFile("output")

        sc.stop()
    }

    /**
      * 自定义分区器
      * 1. 继承Partitioner
      * 2. 重写方法
      */
    class MyPartitioner extends Partitioner{
        // 分区数量
        override def numPartitions: Int = 3

        // 根据数据的key值返回数据所在的分区索引(从0开始)
        override def getPartition(key: Any): Int = {
            key match {
                case "nba" => 0
                case "wnba" => 1
                case _ => 2
            }
        }
    }
}

RDD的保存和加载

Spark 的数据读取及数据保存可以从两个维度来作区分:文件格式以及文件系统。文件格式分为:text 文件、csv 文件、sequence 文件以及Object 文件;
文件系统分为:本地文件系统、HDFS、HBASE 以及数据库。

保存RDD

package ltd.cmjava.spark.core.rdd.io

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

object Spark01_RDD_IO_Save {

    def main(args: Array[String]): Unit = {
        val sparConf = new SparkConf().setMaster("local").setAppName("WordCount")
        val sc = new SparkContext(sparConf)

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

        rdd.saveAsTextFile("output1")
        rdd.saveAsObjectFile("output2")
        rdd.saveAsSequenceFile("output3")

        sc.stop()
    }
}

加载RDD

package ltd.cmjava.spark.core.rdd.io

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

object Spark01_RDD_IO_Load {

    def main(args: Array[String]): Unit = {
        val sparConf = new SparkConf().setMaster("local").setAppName("WordCount")
        val sc = new SparkContext(sparConf)

        val rdd = sc.textFile("output1")
        println(rdd.collect().mkString(","))

        val rdd1 = sc.objectFile[(String, Int)]("output2")
        println(rdd1.collect().mkString(","))

        val rdd2 = sc.sequenceFile[String, Int]("output3")
        println(rdd2.collect().mkString(","))

        sc.stop()
    }
}

累加器

累加器用来把Executor 端变量信息聚合到Driver 端。在Driver 程序中定义的变量,在 Executor 端的每个Task 都会得到这个变量的一份新的副本,每个 task 更新这些副本的值后,传回Driver 端进行 merge。

内置累加器

package ltd.cmjava.spark.core.acc

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

object Spark03_Acc {

    def main(args: Array[String]): Unit = {

        val sparConf = new SparkConf().setMaster("local").setAppName("Acc")
        val sc = new SparkContext(sparConf)

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

        // 获取系统累加器
        // Spark默认就提供了简单数据聚合的累加器
        val sumAcc = sc.longAccumulator("sum")

        //sc.doubleAccumulator
        //sc.collectionAccumulator

        val mapRDD = rdd.map(
            num => {
                // 使用累加器
                sumAcc.add(num)
                num
            }
        )
        println(sumAcc.value) // 少加
        // 获取累加器的值
        // 少加:转换算子中调用累加器,如果没有行动算子的话,那么不会执行
        // 多加:转换算子中调用累加器,如果没有行动算子的话,那么不会执行
        // 一般情况下,累加器会放置在行动算子进行操作
        mapRDD.collect()
        println(sumAcc.value) // 正常
        mapRDD.collect()
        println(sumAcc.value) // 多加

        sc.stop()

    }
}

自定义累加器

package ltd.cmjava.spark.core.acc

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

import scala.collection.mutable

object Spark04_Acc_WordCount {

    def main(args: Array[String]): Unit = {

        val sparConf = new SparkConf().setMaster("local").setAppName("Acc")
        val sc = new SparkContext(sparConf)

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

        // 累加器 : WordCount
        // 创建累加器对象
        val wcAcc = new MyAccumulator()
        // 向Spark进行注册
        sc.register(wcAcc, "wordCountAcc")

        rdd.foreach(
            word => {
                // 数据的累加(使用累加器)
                wcAcc.add(word)
            }
        )

        // 获取累加器累加的结果
        println(wcAcc.value)

        sc.stop()


    }
    /*
      自定义数据累加器:WordCount

      1. 继承AccumulatorV2, 定义泛型
         IN : 累加器输入的数据类型 String
         OUT : 累加器返回的数据类型 mutable.Map[String, Long]

      2. 重写方法(6)
     */
    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合并多个累加器
        override def merge(other: AccumulatorV2[String, mutable.Map[String, Long]]): Unit = {

            val map1 = this.wcMap
            val map2 = other.value

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

        // 累加器结果
        override def value: mutable.Map[String, Long] = {
            wcMap
        }
    }
}

广播变量

如果一个Executor需要运行多个相同的任务,某些变量可能会重复发送,因此需要用到广播变量,让一个Executor获取一次广播变量,共享给多个任务。

package ltd.cmjava.spark.core.acc

import org.apache.spark.broadcast.Broadcast
import org.apache.spark.{SparkConf, SparkContext}

import scala.collection.mutable

object Spark06_Bc {

    def main(args: Array[String]): Unit = {

        val sparConf = new SparkConf().setMaster("local").setAppName("Acc")
        val sc = new SparkContext(sparConf)

        val rdd1 = 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)

        sc.stop()

    }
}
0

评论区