Spark学习笔记(二)

news/2024/3/29 23:02:32/文章来源:https://blog.csdn.net/langzitianya/article/details/128053602

以下笔记基于对尚硅谷spark教程的学习,Spark版本3.0

目录

什么是 RDD

RDD核心属性

执行原理

基础编程

  创建环境

  RDD 创建

    从集合中创建RDD

    从文件中创建RDD

  RDD 转换算子

    Value 类型

    双Value类型

    Key-Value类型

  RDD 行动算子

  RDD 序列化

    闭包检查

    Kryo 序列化框架

  RDD 依赖关系

  RDD 持久化

    RDD Cache 缓存

    RDD CheckPoint 检查点

    缓存和检查点区别

  RDD 分区器

  RDD 文件读取与保存

累加器

  系统累加器

  自定义累加器

  广播变量


什么是 RDD


  RDD(Resilient Distributed Dataset)叫做弹性分布式数据集,是 Spark 中最基本的数据处理模型。代码中是一个抽象类,它代表一个弹性的、不可变、可分区、里面的元素可并行计算的集合。
  ➢ 弹性
    ⚫ 存储的弹性:内存与磁盘的自动切换;
    ⚫ 容错的弹性:数据丢失可以自动恢复;
    ⚫ 计算的弹性:计算出错重试机制;
    ⚫ 分片的弹性:可根据需要重新分片。
  ➢ 分布式:数据存储在大数据集群不同节点上
  ➢ 数据集:RDD 封装了计算逻辑,并不保存数据
  ➢ 数据抽象:RDD 是一个抽象类,需要子类具体实现
  ➢ 不可变:RDD 封装了计算逻辑,是不可以改变的,想要改变,只能产生新的 RDD,在新的 RDD 里面封装计算逻辑
  ➢ 可分区、并行计算

RDD核心属性


  分区列表
    RDD 数据结构中存在分区列表,用于执行任务时并行计算,是实现分布式计算的重要属性。
    protected def getPartitions: Array[Partition]
  分区计算函数
    Spark 在计算时,是使用分区函数对每一个分区进行计算
    @DeveloperApi
    def compute(split: Partition, context: TaskContext): Iterator[T]
  RDD之间的依赖关系
    RDD 是计算模型的封装,当需求中需要将多个计算模型进行组合时,就需要将多个 RDD 建立依赖关系
    protected def getDependencies: Seq[Dependency[_]] = deps
  分区器(可选)
    当数据为 KV 类型数据时,可以通过设定分区器自定义数据的分区
    @transient val partitioner: Option[Partitioner] = None
  首选位置(可选)
    计算数据时,可以根据计算节点的状态选择不同的节点位置进行计算
    protected def getPreferredLocations(split: Partition): Seq[String] = Nil

执行原理


  Spark 框架在执行时,先申请资源,然后将应用程序的数据处理逻辑分解成一个一个的计算任务。然后将任务发到已经分配资源的计算节点上, 按照指定的计算模型进行数据计算。最后得到计算结果。
  在 Yarn 环境中,RDD的工作原理:
    1) 启动 Yarn 集群环境
    2) Spark 通过申请资源创建调度节点和计算节点
    3) Spark 框架根据需求将计算逻辑根据分区划分成不同的任务
    4) 调度节点将任务根据计算节点状态发送到对应的计算节点进行计算

基础编程


  创建环境

val sparkConf = new SparkConf().setMaster("local[*]").setAppName("spark")
val sparkContext = new SparkContext(sparkConf)


  RDD 创建


    从集合中创建RDD

val seq = Seq[Int](1,2,3,4)
// 第二个参数为数据切片/分区的数量,第二个参数不给时,使用默认值defaultParallelism(默认并行度)
//   spark在默认情况下,从配置对象中获取配置参数:spark.default.parallelism
//   如果获取不到,那么使用totalCores属性,这个属性取值为当前运行环境的最大可用核数
val rdd1: RDD[Int] = sparkContext.parallelize(seq, 2)
// makeRDD方法在底层实现时其实就是调用了rdd对象的parallelize方法
val rdd2: RDD[Int] = sparkContext.makeRDD(seq, 2)// 集合读取数据时,每个切片/分区分配的数据地址为
def positions(length: Long, numSlices: Int): Iterator[(Int, Int)] = {(0 until numSlices).iterator.map { i =>val start = ((i * length) / numSlices).toIntval end = (((i + 1) * length) / numSlices).toInt(start, end)}
}

    从文件中创建RDD

// 可以是具体文件,第二个参数为数据切片/分区的最小分区数量,第二个参数不给时,使用默认值math.min(defaultParallelism, 2)
val rdd1: RDD[String] = sparkContext.textFile("datas/1.txt", 3)
// 可以是目录
val rdd2: RDD[String] = sparkContext.textFile("datas")
// 可以使用通配符 *
val rdd3: RDD[String] = sparkContext.textFile("datas/1*.txt")
// 可以是分布式存储系统路径:HDFS
val rdd4: RDD[String] = sparkContext.textFile("hdfs://linux1:8020/test.txt")
// 以文件为单位读取数据,读取的结果表示为元组,第一个元素表示文件路径,第二个元素表示文件内容
val rdd5: RDD[(String, String)] = sparkContext.wholeTextFiles("datas")

      读取文件数据时,数据是按照 Hadoop 文件读取的规则进行切片分区
      文件读取数据时实际切片/分区数 >=最小分区数
      单个文件时
        分区数的计算方式:
          每个分区字节数 = 文件总字节数 / 最小分区数
          分区数 ≈ 文件总字节数 / 每个分区字节数
            余数部分>每个分区字节数*0.1时,分区数+1
        数据分区的划分:
          1.数据以行为单位进行读取,一行数据只会属于一个分区
          2.数据分区偏移量计算
            例如:文件总字节数为7,最小分区数设为2,每个分区的字节数为3,实际分区数为3
              分区1的偏移量: [0, 3]
              分区2的偏移量: [3, 6]
              分区3的偏移量: [7]
          3.以每行数据的第一个字节的偏移量来决定属于哪个分区
      如果数据源为多个文件,那么计算分区时以文件为单位进行分区


  RDD 转换算子


    RDD 根据数据处理方式的不同将算子整体上分为 Value 类型、双 Value 类型和 Key-Value类型


    Value 类型


      map 一进一出

// 将处理的数据逐条进行映射转换,这里的转换可以是类型的转换,也可以是值的转换。
def map[U](f: (T) ⇒ U)(implicit arg0: ClassTag[U]): RDD[U]// 1. rdd的计算一个分区内的数据是一个一个执行逻辑//    只有前面一个数据全部的逻辑执行完毕后,才会执行下一个数据。//    分区内数据的执行是有序的。// 2. 不同分区数据计算是无序的。

 
      mapPartitions 多进多出

// 将待处理的数据以分区为单位发送到计算节点进行处理,这里的处理是指可以进行任意的处理,哪怕是过滤数据。
def mapPartitions[U](f: (Iterator[T]) ⇒ Iterator[U], preservesPartitioning: Boolean = false)(implicit arg0: ClassTag[U]): RDD[U]

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

      mapPartitionsWithIndex 多进多出

// 将待处理的数据以分区为单位发送到计算节点进行处理,这里的处理是指可以进行任意的处理,哪怕是过滤数据,在处理时同时可以获取当前分区索引。
def mapPartitionsWithIndex[U](f: (Int, Iterator[T]) ⇒ Iterator[U], preservesPartitioning: Boolean = false)(implicit arg0: ClassTag[U]): RDD[U]

      flatMap 一进多出

// 将处理的数据进行扁平化后再进行映射处理,所以算子也称之为扁平映射
def flatMap[U](f: (T) ⇒ TraversableOnce[U])(implicit arg0: ClassTag[U]): RDD[U]

      glom

// 将同一个分区的数据直接转换为相同类型的内存数组进行处理,分区不变
def glom(): RDD[Array[T]]

      groupBy

// 将数据根据指定的规则进行分组, 分区数量默认不变,但是数据会被打乱重新组合,我们将这样的操作称之为 shuffle。
// 极限情况下,数据可能被分在同一个分区中,一个组的数据在一个分区中,但是并不是说一个分区中只有一个组
def groupBy[K](f: (T) ⇒ K)(implicit kt: ClassTag[K]): RDD[(K, Iterable[T])]

      filter

// 将数据根据指定的规则进行筛选过滤,符合规则的数据保留,不符合规则的数据丢弃。
// 当数据进行筛选过滤后,分区不变,但是分区内的数据可能不均衡,生产环境下,可能会出现数据倾斜。
def filter(f: (T) ⇒ Boolean): RDD[T]

      sample

// 根据指定的规则从数据集中抽取数据
def sample(withReplacement: Boolean, fraction: Double, seed: Long = Utils.random.nextLong): RDD[T]// 抽取数据不放回(伯努利算法)// 第一个参数:抽取的数据是否放回,false:不放回// 第二个参数:抽取的几率,范围在[0,1]之间,0:全不取;1:全取// 第三个参数:随机数种子// 抽取数据放回(泊松算法)// 第一个参数:抽取的数据是否放回,true:放回// 第二个参数:重复数据的几率,范围大于等于0.表示每一个元素被期望抽取到的次数// 第三个参数:随机数种子

      distinct

// 将数据集中重复的数据去重,如果传入参数numPartitions,表示去重后的分区数
// 在未指定分区器partitioner时,实现方式是:map(x => (x, null)).reduceByKey((x, _) => x, numPartitions).map(_._1)
def distinct(): RDD[T]
def distinct(numPartitions: Int)(implicit ord: Ordering[T] = null): RDD[T]

      coalesce

// 根据数据量缩减分区,用于大数据集过滤后,提高小数据集的执行效率
// 默认情况下不会将分区的数据打乱重新组合,这种情况下的缩减分区可能会导致数据不均衡,出现数据倾斜,如果想要让数据均衡,可以进行shuffle处理
// coalesce算子可以扩大分区的,但是如果不进行shuffle操作,是没有意义,不起作用。所以如果想要实现扩大分区的效果,需要使用shuffle操作
def coalesce(numPartitions: Int, shuffle: Boolean = false, partitionCoalescer: Option[PartitionCoalescer] = Option.empty)(implicit ord: Ordering[T] = null): RDD[T]

      repartition

// 该操作内部其实执行的是 coalesce 操作,参数 shuffle 的默认值为 true。
// 无论是将分区数多的RDD 转换为分区数少的 RDD,还是将分区数少的 RDD 转换为分区数多的 RDD,repartition操作都可以完成,因为无论如何都会经 shuffle 过程。
def repartition(numPartitions: Int)(implicit ord: Ordering[T] = null): RDD[T]

      sortBy

// 该操作用于排序数据。在排序之前,可以将数据通过 f 函数进行处理,之后按照 f 函数处理的结果进行排序,默认为升序排列。
// 排序后新产生的 RDD 的分区数与原 RDD 的分区数一致。中间存在 shuffle 的过程
def sortBy[K](f: (T) ⇒ K, ascending: Boolean = true, numPartitions: Int = this.partitions.length)(implicit ord: Ordering[K], ctag: ClassTag[K]): RDD[T]

    双Value类型


      intersection

// 对源 RDD 和参数 RDD 求交集后返回一个新的 RDD,要求数据类型一致
def intersection(other: RDD[T], numPartitions: Int): RDD[T]
def intersection(other: RDD[T], partitioner: Partitioner)(implicit ord: Ordering[T] = null): RDD[T]
def intersection(other: RDD[T]): RDD[T]

      union

// 对源 RDD 和参数 RDD 求并集后返回一个新的 RDD,要求数据类型一致
def union(other: RDD[T]): RDD[T]

      subtract

// 以一个 RDD 元素为主,去除两个 RDD 中重复元素,将其他元素保留下来。求差集,要求数据类型一致
def subtract(other: RDD[T], p: Partitioner)(implicit ord: Ordering[T] = null): RDD[T]
def subtract(other: RDD[T], numPartitions: Int): RDD[T]
def subtract(other: RDD[T]): RDD[T]

      zip

// 拉链,将两个 RDD 中的元素,以键值对的形式进行合并。其中,键值对中的 Key 为第 1 个 RDD中的元素,Value 为第 2 个 RDD 中的相同位置的元素。
// 要求分区数量相等,每个分区中的数据个数相等
def zip[U](other: RDD[U])(implicit arg0: ClassTag[U]): RDD[(T, U)]

    Key-Value类型


      partitionBy

// 类PairRDDFunctions的方法,存在RDD->PairRDDFunctions的隐式转换
// 将数据按照指定 Partitioner 重新进行分区。Spark 默认的分区器是 HashPartitioner
def partitionBy(partitioner: Partitioner): RDD[(K, V)]// 如果重分区的分区器和当前 RDD 的分区器一样,什么都不做

      reduceByKey

// 类PairRDDFunctions的方法,存在RDD->PairRDDFunctions的隐式转换
// 将数据按照相同的 Key 对 Value 进行聚合
// scala语言中一般的聚合操作都是两两聚合,spark基于scala开发的,所以它的聚合也是两两聚合
// reduceByKey中如果key的数据只有一个,是不会参与运算的。
def reduceByKey(func: (V, V) ⇒ V): RDD[(K, V)]
// 需要重分区的时候
def reduceByKey(func: (V, V) ⇒ V, numPartitions: Int): RDD[(K, V)]
def reduceByKey(partitioner: Partitioner, func: (V, V) ⇒ V): RDD[(K, V)]val dataRDD1 = sparkContext.makeRDD(List(("a",1),("b",2),("c",3)))
val dataRDD2 = dataRDD1.reduceByKey(_+_)
val dataRDD3 = dataRDD1.reduceByKey(_+_, 2)

      groupByKey

// 类PairRDDFunctions的方法,存在RDD->PairRDDFunctions的隐式转换
// 将数据源中相同key的数据分在一个组中,形成一个对偶元组,元组中的第一个元素就是key,第二个元素就是相同key的value的集合
def groupByKey(): RDD[(K, Iterable[V])]
// 需要重分区的时候
def groupByKey(numPartitions: Int): RDD[(K, Iterable[V])]
def groupByKey(partitioner: Partitioner): RDD[(K, Iterable[V])]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 的区别
        从 shuffle 的角度:
           reduceByKey 和 groupByKey 都存在 shuffle 的操作,都会落盘。
           但是reduceByKey可以在 shuffle 前对分区内相同 key 的数据进行预聚合(combine)功能,这样会减少落盘的数据量,
           而 groupByKey 只是进行分组,不存在数据量减少的问题,reduceByKey 性能比较高。
        从功能的角度:
           reduceByKey 其实包含分组和聚合的功能。groupByKey 只能分组,不能聚合。
           所以在分组聚合的场合下,推荐使用 reduceByKey,
           如果仅仅是分组而不需要聚合,那么还是只能使用 groupByKey

      aggregateByKey

// 类PairRDDFunctions的方法,存在RDD->PairRDDFunctions的隐式转换
// 将数据根据不同的规则进行分区内计算和分区间计算
def aggregateByKey[U](zeroValue: U)(seqOp: (U, V) ⇒ U, combOp: (U, U) ⇒ U)(implicit arg0: ClassTag[U]): RDD[(K, U)]// aggregateByKey存在函数柯里化,有两个参数列表// 1. 第一个参数列表中的参数表示初始值//   主要用于当碰见第一个key的时候,和value进行分区内计算// 2. 第二个参数列表中含有两个参数//   2.1 第一个参数表示分区内的计算规则//   2.2 第二个参数表示分区间的计算规则
// 需要重分区的时候
def aggregateByKey[U](zeroValue: U, numPartitions: Int)(seqOp: (U, V) ⇒ U, combOp: (U, U) ⇒ U)(implicit arg0: ClassTag[U]): RDD[(K, U)]
def aggregateByKey[U](zeroValue: U, partitioner: Partitioner)(seqOp: (U, V) ⇒ U, combOp: (U, U) ⇒ U)(implicit arg0: ClassTag[U]): RDD[(K, U)]val rdd = sc.makeRDD(List(("a", 1), ("a", 2), ("a", 3), ("a", 4)
),2)
rdd.aggregateByKey(0)((x, y) => math.max(x, y),(x, y) => x + y
).collect.foreach(println)

      foldByKey

// 类PairRDDFunctions的方法,存在RDD->PairRDDFunctions的隐式转换
// 当分区内计算规则和分区间计算规则相同时,aggregateByKey 就可以简化为 foldByKey
def foldByKey(zeroValue: V)(func: (V, V) ⇒ V): RDD[(K, V)]
// 需要重分区的时候
def foldByKey(zeroValue: V, numPartitions: Int)(func: (V, V) ⇒ V): RDD[(K, V)]
def foldByKey(zeroValue: V, partitioner: Partitioner)(func: (V, V) ⇒ V): RDD[(K, V)]

      combineByKey

// 类PairRDDFunctions的方法,存在RDD->PairRDDFunctions的隐式转换
// 最通用的对 key-value 型 rdd 进行聚集操作的聚集函数(aggregation function)。类似于aggregate(),combineByKey()允许用户返回值的类型与输入不一致。
def combineByKey[C](createCombiner: (V) ⇒ C, mergeValue: (C, V) ⇒ C, mergeCombiners: (C, C) ⇒ C): RDD[(K, C)]// 第一个参数表示:将相同key的第一个数据进行结构的转换,实现操作// 第二个参数表示:分区内的计算规则// 第三个参数表示:分区间的计算规则
// 需要重分区的时候
def combineByKey[C](createCombiner: (V) ⇒ C, mergeValue: (C, V) ⇒ C, mergeCombiners: (C, C) ⇒ C, numPartitions: Int): RDD[(K, C)]
def combineByKey[C](createCombiner: (V) ⇒ C, mergeValue: (C, V) ⇒ C, mergeCombiners: (C, C) ⇒ C, partitioner: Partitioner, mapSideCombine: Boolean = true, serializer: Serializer = null): RDD[(K, C)]
def combineByKeyWithClassTag[C](createCombiner: (V) ⇒ C, mergeValue: (C, V) ⇒ C, mergeCombiners: (C, C) ⇒ C)(implicit ct: ClassTag[C]): RDD[(K, C)]

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

        四个函数底层都是通过调用下面的函数来实现的:
          def combineByKeyWithClassTag[C](createCombiner: (V) ⇒ C, mergeValue: (C, V) ⇒ C, mergeCombiners: (C, C) ⇒ C, partitioner: Partitioner, mapSideCombine: Boolean = true, serializer: Serializer = null)(implicit ct: ClassTag[C]): RDD[(K, C)]

      join

// 类PairRDDFunctions的方法,存在RDD->PairRDDFunctions的隐式转换
// 在类型为(K,V)和(K,W)的 RDD 上调用,返回一个相同 key 对应的所有元素连接在一起的(K,(V,W))的 RDD
def join[W](other: RDD[(K, W)]): RDD[(K, (V, W))]
// 需要重分区的时候
def join[W](other: RDD[(K, W)], numPartitions: Int): RDD[(K, (V, W))]
def join[W](other: RDD[(K, W)], partitioner: Partitioner): RDD[(K, (V, W))]

      leftOuterJoin/rightOuterJoin/fullOuterJoin

// 类PairRDDFunctions的方法,存在RDD->PairRDDFunctions的隐式转换
// 类似于 SQL 语句的左外连接/右外连接/全外连接
def leftOuterJoin[W](other: RDD[(K, W)]): RDD[(K, (V, Option[W]))]
def rightOuterJoin[W](other: RDD[(K, W)]): RDD[(K, (Option[V], W))]
def fullOuterJoin[W](other: RDD[(K, W)]): RDD[(K, (Option[V], Option[W]))]
// 需要重分区的时候
def leftOuterJoin[W](other: RDD[(K, W)], numPartitions: Int): RDD[(K, (V, Option[W]))]
def leftOuterJoin[W](other: RDD[(K, W)], partitioner: Partitioner): RDD[(K, (V, Option[W]))]
def rightOuterJoin[W](other: RDD[(K, W)], numPartitions: Int): RDD[(K, (Option[V], W))]
def rightOuterJoin[W](other: RDD[(K, W)], partitioner: Partitioner): RDD[(K, (Option[V], W))]
def fullOuterJoin[W](other: RDD[(K, W)], numPartitions: Int): RDD[(K, (Option[V], Option[W]))]
def fullOuterJoin[W](other: RDD[(K, W)], partitioner: Partitioner): RDD[(K, (Option[V], Option[W]))]

      cogroup
 

// 类PairRDDFunctions的方法,存在RDD->PairRDDFunctions的隐式转换
// connect + group,在类型为(K,V)和(K,W)的 RDD 上调用,返回一个(K,(Iterable<V>,Iterable<W>))类型的 RDD
def cogroup[W](other: RDD[(K, W)]): RDD[(K, (Iterable[V], Iterable[W]))]
def cogroup[W1, W2](other1: RDD[(K, W1)], other2: RDD[(K, W2)]): RDD[(K, (Iterable[V], Iterable[W1], Iterable[W2]))]
def cogroup[W1, W2, W3](other1: RDD[(K, W1)], other2: RDD[(K, W2)], other3: RDD[(K, W3)]): RDD[(K, (Iterable[V], Iterable[W1], Iterable[W2], Iterable[W3]))]
// 需要重分区的时候
def cogroup[W](other: RDD[(K, W)], numPartitions: Int): RDD[(K, (Iterable[V], Iterable[W]))]
def cogroup[W1, W2](other1: RDD[(K, W1)], other2: RDD[(K, W2)], numPartitions: Int): RDD[(K, (Iterable[V], Iterable[W1], Iterable[W2]))]
def cogroup[W1, W2, W3](other1: RDD[(K, W1)], other2: RDD[(K, W2)], other3: RDD[(K, W3)], numPartitions: Int): RDD[(K, (Iterable[V], Iterable[W1], Iterable[W2], Iterable[W3]))]
def cogroup[W](other: RDD[(K, W)], partitioner: Partitioner): RDD[(K, (Iterable[V], Iterable[W]))]
def cogroup[W1, W2](other1: RDD[(K, W1)], other2: RDD[(K, W2)], partitioner: Partitioner): RDD[(K, (Iterable[V], Iterable[W1], Iterable[W2]))]
def cogroup[W1, W2, W3](other1: RDD[(K, W1)], other2: RDD[(K, W2)], other3: RDD[(K, W3)], partitioner: Partitioner): RDD[(K, (Iterable[V], Iterable[W1], Iterable[W2], Iterable[W3]))]

  RDD 行动算子


    所谓的行动算子,其实就是触发作业(Job)执行的方法

    reduce

// 聚集 RDD 中的所有元素,先聚合分区内数据,再聚合分区间数据
def reduce(f: (T, T) ⇒ T): T

    collect

// 将不同分区的数据按照分区顺序采集到Driver端内存中,形成数组
def collect(): Array[T]
// 收集的同时做类型转换
def collect[U](f: PartialFunction[T, U])(implicit arg0: ClassTag[U]): RDD[U]

    count

// 返回 RDD 中元素的个数
def count(): Long

    first

// 返回 RDD 中的第一个元素
def first(): T

    take

// 返回一个由 RDD 的前 n 个元素组成的数组
def take(num: Int): Array[T]

    takeOrdered

// 返回该 RDD 排序后的前 n 个元素组成的数组,默认升序,可以通过传入第二个参数自定义排序
def takeOrdered(num: Int)(implicit ord: Ordering[T]): Array[T]

    takeSample

// 随机抽样算子
// withReplacement:是否可以多次抽样(true可以,false不可以)
// num:返回的样本的大小
// seed:随机数生成器的种子(一般都是默认不会指定)
def takeSample(withReplacement: Boolean, num: Int, seed: Long = Utils.random.nextLong): Array[T]

    top

// 返回该 RDD 排序后的前 n 个元素组成的数组,默认降序,可以通过传入第二个参数自定义排序
def top(num: Int)(implicit ord: Ordering[T]): Array[T]

    aggregate

// 分区的数据通过初始值和分区内的数据进行聚合,然后再和初始值进行分区间的数据聚合
//   不同:aggregateByKey的初始值只参与分区内计算
def aggregate[U](zeroValue: U)(seqOp: (U, T) ⇒ U, combOp: (U, U) ⇒ U)(implicit arg0: ClassTag[U]): U

    fold

// 折叠操作,aggregate 的简化版操作
def fold(zeroValue: T)(op: (T, T) ⇒ T): T

    countByKey

// 类PairRDDFunctions的方法,存在RDD->PairRDDFunctions的隐式转换
// 统计每种 key 的个数
def countByKey(): Map[K, Long]

    countByValue

// 统计每种 value 的个数
def countByValue()(implicit ord: Ordering[T] = null): Map[T, Long]

    save 相关算子

// 将数据保存到不同格式的文件中,每个分区生成一个文件
def saveAsTextFile(path: String): Unit
def saveAsTextFile(path: String, codec: Class[_ <: CompressionCodec]): Unit
def saveAsObjectFile(path: String): Unit
// 类SequenceFileRDDFunctions的方法,存在RDD->SequenceFileRDDFunctions的隐式转换,数据必须为(key, value)类型
def saveAsSequenceFile(path: String, codec: Option[Class[_ <: CompressionCodec]] = None): Unit

    foreach

// 分布式遍历 RDD 中的每一个元素,调用指定函数
def foreach(f: (T) ⇒ Unit): Unit// 这里的foreach其实是Scala的Array的方法,在Driver端内存集合的循环遍历方法
rdd.collect().foreach(println)// 这里的foreach其实是Executor端内存数据打印
rdd.foreach(println)

  RDD 序列化


    闭包检查


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

    Kryo 序列化框架


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

// 设定其他类型也使用Kryo序列化
val conf: SparkConf = new SparkConf().setAppName("SerDemo").setMaster("local[*]")// 替换默认的序列化机制.set("spark.serializer", "org.apache.spark.serializer.KryoSerializer")// 注册需要使用 kryo 序列化的自定义类.registerKryoClasses(Array(classOf[Searcher]))

  RDD 依赖关系


    RDD 血缘关系
      将创建 RDD 的一系列 Lineage(血统)记录下来,以便恢复丢失的分区。
      RDD 的 Lineage 会记录 RDD 的元数据信息和转换行为,当该 RDD 的部分分区数据丢失时,它可以根据这些信息来重新运算和恢复丢失的数据分区。
    RDD 依赖关系
      这里所谓的依赖关系,其实就是两个相邻 RDD 之间的关系
    RDD 窄依赖
      窄依赖表示每一个父(上游)RDD 的 Partition 最多被子(下游)RDD 的一个 Partition 使用,窄依赖我们形象的比喻为独生子女。
        class OneToOneDependency[T] extends NarrowDependency[T]
    RDD 宽依赖
      宽依赖表示同一个父(上游)RDD 的 Partition 被多个子(下游)RDD 的 Partition 依赖,会引起 Shuffle,宽依赖我们形象的比喻为多生。
        class ShuffleDependency[K, V, C] extends Dependency[Product2[K, V]]
    RDD 阶段划分
      RDD中存在宽依赖(ShuffleDependency)时,阶段会增加一个,阶段的数量 = 宽依赖(ShuffleDependency)的数量 + 1
    RDD 任务划分
      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中不存储数据,默认情况下如果一个RDD需要重复使用,那么需要从头执行来获取数据

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

      缓存有可能丢失,或者存储于内存的数据由于内存不足而被删除,RDD 的缓存容错机制保证了即使缓存丢失也能保证计算的正确执行。
      通过基于 RDD 的一系列转换,丢失的数据会被重算,由于 RDD 的各个 Partition 是相对独立的,因此只需要计算丢失的部分即可,并不需要重算全部 Partition。

      Spark 会自动对一些 Shuffle 操作的中间数据做持久化操作(比如:reduceByKey)。这样做的目的是为了当一个节点 Shuffle 失败了避免重新计算整个输入。
      但是,在实际使用的时候,如果想重用数据,仍然建议调用 persist 或 cache。

// cache默认持久化的操作,只能将数据保存到内存中,如果想要保存到磁盘文件,需要更改存储级别
//mapRDD.cache()
// 持久化操作必须在行动算子执行时完成的。
mapRDD.persist(StorageLevel.DISK_ONLY)mapRDD.reduceByKey(_+_).collect().foreach(println)
// 如果前面mapRDD没有调用cache或者persist方法,第二次使用时会从头执行来获取数据
mapRDD.groupByKey().collect().foreach(println)

    RDD CheckPoint 检查点


      所谓的检查点其实就是通过将 RDD 中间结果写入磁盘
      由于血缘依赖过长会造成容错成本过高,这样就不如在中间阶段做检查点容错,如果检查点之后有节点出现问题,可以从检查点开始重做血缘,减少了开销。
      对 RDD 进行 checkpoint 操作并不会马上被执行,必须执行 Action 操作才能触发。
      
      检查点路径保存的文件,当作业执行完毕后,不会被删除,一般保存路径都是在分布式存储系统:HDFS

// 设置检查点路径
sc.setCheckpointDir("./checkpoint1")
// 创建一个 RDD,读取指定位置文件:hello atguigu atguigu
val lineRdd: RDD[String] = sc.textFile("input/1.txt")
// 业务逻辑
val wordRdd: RDD[String] = lineRdd.flatMap(line => line.split(" "))
val wordToOneRdd: RDD[(String, Long)] = wordRdd.map {word => {(word, System.currentTimeMillis())}
}
// 增加缓存,避免再重新跑一个 job 做 checkpoint
wordToOneRdd.cache()
// 数据检查点:针对 wordToOneRdd 做检查点计算
wordToOneRdd.checkpoint()
// 触发执行逻辑
wordToOneRdd.collect().foreach(println)

    缓存和检查点区别


      cache : 将数据临时存储在内存中进行数据重用
              会在血缘关系中添加新的依赖。一旦,出现问题,可以重头读取数据
      persist : 将数据临时存储在磁盘文件中进行数据重用
                涉及到磁盘IO,性能较低,但是数据安全
                如果作业执行完毕,临时保存的数据文件就会丢失
      checkpoint : 将数据长久地保存在磁盘文件中进行数据重用
                涉及到磁盘IO,性能较低,但是数据安全
                为了保证数据安全,所以一般情况下,会独立执行作业
                为了能够提高效率,一般情况下,是需要和cache联合使用
                执行过程中,会切断血缘关系。重新建立新的血缘关系
                checkpoint等同于改变数据源

  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 分区:将一定范围内的数据映射到一个分区中,尽量保证每个分区数据均匀,而且分区间有序

  RDD 文件读取与保存


    文件格式分为:text 文件、csv 文件、sequence 文件以及 Object 文件
    文件系统分为:本地文件系统、HDFS、HBASE 以及数据库

    text 文件

// 读取输入文件
val inputRDD: RDD[String] = sc.textFile("input/1.txt")
// 保存数据
inputRDD.saveAsTextFile("output")

    sequence 文件
      SequenceFile 文件是 Hadoop 用来存储二进制形式的 key-value 对而设计的一种平面文件(Flat File)

// 保存数据为 SequenceFile
dataRDD.saveAsSequenceFile("output")
// 读取 SequenceFile 文件
sc.sequenceFile[Int,Int]("output").collect().foreach(println)

    object 对象文件
      对象文件是将对象序列化后保存的文件,采用 Java 的序列化机制。

// 保存数据
dataRDD.saveAsObjectFile("output")
// 读取数据
sc.objectFile[Int]("output").collect().foreach(println)

累加器


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

  系统累加器

    val rdd = sc.makeRDD(List(1,2,3,4,5))// 声明累加器var sum = sc.longAccumulator("sum");rdd.foreach(num => {// 使用累加器sum.add(num)})// 获取累加器的值println("sum = " + sum.value)

  自定义累加器

    // 自定义累加器// 1. 继承 AccumulatorV2,并设定输入和输出泛型// 2. 重写累加器的抽象方法class WordCountAccumulator extends AccumulatorV2[String, mutable.Map[String, Long]]{var map : mutable.Map[String, Long] = mutable.Map()// 累加器是否为初始状态override def isZero: Boolean = {map.isEmpty}// 复制累加器override def copy(): AccumulatorV2[String, mutable.Map[String, Long]] = {new WordCountAccumulator}// 重置累加器override def reset(): Unit = {map.clear()}// 向累加器中增加数据 (In)override def add(word: String): Unit = {// 查询 map 中是否存在相同的单词// 如果有相同的单词,那么单词的数量加 1// 如果没有相同的单词,那么在 map 中增加这个单词map(word) = map.getOrElse(word, 0L) + 1L}// 合并累加器override def merge(other: AccumulatorV2[String, mutable.Map[String, Long]]): Unit = {val map1 = mapval map2 = other.value// 两个 Map 的合并map = map1.foldLeft(map2)(( innerMap, kv ) => {innerMap(kv._1) = innerMap.getOrElse(kv._1, 0L) + kv._2innerMap})}// 返回累加器的结果 (Out)override def value: mutable.Map[String, Long] = map}// 使用自定义累加器// 创建累加器对象val wcAcc = new WordCountAccumulator()// 向Spark进行注册sc.register(wcAcc, "wordCountAcc")rdd.foreach(word => {// 数据的累加(使用累加器)wcAcc.add(word)})// 获取累加器累加的结果println(wcAcc.value)

  广播变量


    广播变量用来高效分发较大的对象。向所有工作节点(Executor)发送一个较大的只读值,以供一个或多个 Spark 操作使用。
    在多个并行操作中使用同一个变量,但是 Spark 会为每个任务分别发送。

    val rdd1 = sc.makeRDD(List( ("a",1), ("b", 2), ("c", 3), ("d", 4) ),4)val list = List( ("a",4), ("b", 5), ("c", 6), ("d", 7) )// 声明广播变量val broadcast: Broadcast[List[(String, Int)]] = sc.broadcast(list)val resultRDD: RDD[(String, (Int, Int))] = rdd1.map {case (key, num) => {var num2 = 0// 使用广播变量for ((k, v) <- broadcast.value) {if (k == key) {num2 = v}}(key, (num, num2))}}

本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如若转载,请注明出处:http://www.luyixian.cn/news_show_37252.aspx

如若内容造成侵权/违法违规/事实不符,请联系dt猫网进行投诉反馈email:809451989@qq.com,一经查实,立即删除!

相关文章

【JVM技术专题】「原理专题」全流程分析Java对象的创建过程及内存布局

前言概要 对应过程则是&#xff1a;对象创建、对象内存布局、对象访问定位的三个过程。 对象的创建过程 对象的创建方式 java中对象的创建方式有很多种&#xff0c;常见的是通过new关键字和反射这两种方式来创建。除此之外&#xff0c;还有clone、反序列化等方式创建。 通过n…

音乐播放

在Qt5中使用Qt Multimedia 模块来实现多媒体应用&#xff08;音视频播放和控制&#xff0c;相机拍照。收音等&#xff09;。 使用多媒体模块时需要在pro文件中添加&#xff1a; QT multimedia QMediaPlayer&#xff08;播放音频&#xff09; 不追求低延迟的话使用QMediaPlaye…

招生CRM系统|基于Springboot实现培训机构招生CRM管理系统

作者主页&#xff1a;编程指南针 作者简介&#xff1a;Java领域优质创作者、CSDN博客专家 、掘金特邀作者、多年架构师设计经验、腾讯课堂常驻讲师 主要内容&#xff1a;Java项目、毕业设计、简历模板、学习资料、面试题库、技术互助 收藏点赞不迷路 关注作者有好处 文末获取源…

JavaWeb开发之——多表查询(21)

一 概述 多表查询—简介多表查询—内连接&外连接多表查询—子查询 二 多表查询—简介 2.1 概念 多表查询顾名思义就是从多张表中一次性的查询出我们想要的数据。 2.2 SQL数据准备 DROP TABLE IF EXISTS emp; DROP TABLE IF EXISTS dept; # 创建部门表 CREATE TABLE de…

UE5笔记【九】蓝图BluePrint;

新建一个第三视角游戏。然后打开关卡蓝图。 长得跟材料编辑器一样。 这里是我们创建Node和新功能的地方。 首先我们新建一个游戏开始的地方。右键&#xff1a;Begin搜索。 我们需要打印一行字&#xff1a;欢迎来到游戏世界。我们需要添加一个打印文本的结点&#xff1a;PrintT…

FilterListenerAjax的介绍

目录 一、Filter 1、Filter概述 2、过滤器链 二、Listener 三、Ajax 1、基本介绍 2、快速入门案例 3、axios 4、JSON 一、Filter 1、Filter概述 ▶ 过滤器 Filter 表示过滤器&#xff0c;是 JavaWeb 三大组件(Servlet、Filter、Listener)之一。过滤器可以把对资源的请…

论文阅读: Disentangled lmage Colorization via Global Anchors

Disentangled lmage Colorization via Global Anchors发表于SIGGRAPH ASIA 2022&#xff0c;是一篇基于深度学习的图像彩色化的工作&#xff0c;简单介绍一下。之前曾分享过一篇彩色化的经典论文&#xff1a;经典论文回顾: Colorization using Optimization。 作者认为图像彩色…

Spring Cloud OpenFeign - 日志配置

项目源码地址&#xff1a;https://download.csdn.net/download/weixin_42950079/87168704 OpenFeign 有 4 种日志级别&#xff1a; NONE: 不记录任何日志&#xff0c;是OpenFeign默认日志级别&#xff08;性能最佳&#xff0c;适用于生产环境&#xff09;。BASIC: 仅记录请求方…

【机器学习入门项目10例】(八):贝叶斯网络-拼写检查器

💥 项目专栏:【机器学习入门项目10例】 文章目录 一、贝叶斯网络-拼写检查器二、数据集介绍三、导包四、读取数据集五、构建拼写检查器5.1 分词操作5.2 计算词频5.3 计算概率5.4 调用入口5.5 贝叶斯网络模型类六、模型测试一、贝叶斯网络-拼写检查器 1)建立一个足够大的文本…

让学前端不再害怕英语单词(四)

|| 欢迎关注csdn前端领域博主: 前端小王hs || email: 337674757qq.com || 前端交流群&#xff1a; 598778642前三章直通车↓↓↓ 让学前端不再害怕英语单词&#xff08;一&#xff09; 让学前端不再害怕英语单词&#xff08;二&#xff09; 让学前端不再害怕英语单词&#xff0…

Java - 利用Nacos做一个动态开关配置功能

Java - 利用Nacos做一个动态开关配置功能前言一. Nacos配置类编写二. 测试三. 展望前言 我公司里有一个Config配置功能&#xff08;我相信这是很普遍的一个功能&#xff09;。简单来说就是&#xff1a; 将相关的键值对放到这个Config配置系统里面。代码里通过这个Config配置系…

【C++中预处理语句 include、define、if】

1.预处理阶段 预处理阶段&#xff0c;在实际发生编译之前就根据对应的预处理语句进行操作&#xff0c;等到预处理阶段完成之后才进行编译阶段 。 2.预处理语句 预处理语句主要有include、define、if 和 program。利用 # 进行标记 2.1 include语句 include语句就是将所包含的…

【点云处理】点云法向量估计及其加速(4)

上篇文章【点云处理】点云法向量估计及其加速(3)介绍了如何使用pcl提供的gpu版本法向量计算接口对点云发向量计算进行加速。不足之处在于点云k近邻查找依然比较耗时&#xff0c;成为影响整体计算性能的瓶颈。这篇文章就如何优化点云K近邻查找效率进行实验。上一篇文章的示例代码…

想要精通算法和SQL的成长之路 - 无重叠区间

想要精通算法和SQL的成长之路 - 无重叠区间前言一. 无重叠区间前言 想要精通算法和SQL的成长之路 - 系列导航 一. 无重叠区间 原题链接 给定一个区间的集合 intervals &#xff0c;其中 intervals[i] [starti, endi] 。返回 需要移除区间的最小数量&#xff0c;使剩余区间互…

谷粒学苑_第十天

第十天 视频删除 后端 相关sdk在阿里云视频点播文档的服务端SDK–>Java SDK–>媒资管理–>删除视频 复制前面的InitObject到utils里 删除的方法 DeleteMapping("{id}")public R removeAliyunVideo(PathVariable String id){try{DefaultAcsClient defau…

1000套web前端期末大作业 HTML+CSS+JavaScript网页设计实例 企业网站制作【建议收藏】

一、1000套HTML期末学生结课大作业作品(HTMLCSSJS) 这8年来做了1000多套(HTMLCSSJS)网页设计的学生期末大作业&#xff0c;都是给学生定制的都符合学校或者学生考试期末作业的水平&#xff0c;都是divcss框架原创代码写的&#xff0c;有的有js&#xff0c;有的视频音乐flash的…

Mongodb操作基础 分片

Mongodb分片 MongoDB分片是MongoDB支持的另一种集群形式&#xff0c;它可以满足MongoDB数据量呈爆发式增长的需求。当MongoDB存储海量的数据时&#xff0c;一台机器可能无法满足数据存储的需求&#xff0c;也可能无法提供可接受的读写吞吐量&#xff0c;这时&#xff0c;我们就…

【算法】2022第五届“传智杯”全国大学生计算机大赛(练习赛)

【参考&#xff1a;第五届“传智杯”全国大学生计算机大赛&#xff08;练习赛&#xff09; - 洛谷 | 计算机科学教育新生态】 练习赛满分程序&#xff08;多语言&#xff09;&#xff1a;https://www.luogu.com.cn/paste/fi60s4yu CPU一秒大概运行 10810^8108 次&#xff0c;…

ASEMI肖特基二极管MBR40200PT参数,MBR40200PT规格

编辑-Z ASEMI肖特基二极管MBR40200PT参数&#xff1a; 型号&#xff1a;MBR40200PT 最大重复峰值反向电压&#xff08;VRRM&#xff09;&#xff1a;200V 最大平均正向整流输出电流&#xff08;IF&#xff09;&#xff1a;40A 峰值正向浪涌电流&#xff08;IFSM&#xff0…

imx6ull pro BSP 工具链

BSP&#xff0c;Board Support Package&#xff0c;指板级支持包&#xff0c;是构建嵌入式操作系统所 需的引导程序(Bootload)、内核(Kernel)、根文件系统(Rootfs)和工具链 (Toolchain)。 每种开发板的 BSP 都不一样&#xff0c;并且这些源码都非常庞大。我们把这些源码都 放在…