[大数据]大数据——Spark RDD常用算子总结

Spark的核心是建立在同一的抽象弹性分布式数据集（Resilient Distributed Datasets，RDD）之上的，这使得Spark的各个组件可以无缝的进行集成，能够在同一个应用程序中完成大数据处理

1.RDD基本概念

RDD是spark提供的最重要的抽象概念，它是一种有容错机制的特殊数据集合，可以分布在集群的节点上，以函数式操作集合的方式进行各种并行操作

可以将RDD理解为一个分布式对象集合，本质上是一个只读的分区记录集合。每个RDD可以分成多个分区，每个分区就是一个数据集片段。一个RDD的不同分区可以保存到集群中的不同节点上，从而可以在集群中的不同节点上进行并行计算。

弹性：

????????存储的弹性：内存与磁盘的自动切换

????????容错的弹性：数据丢失可以自动恢复

????????计算的弹性：计算出错重试机制

????????分片的弹性：可以根据需要重新分片

分布式：

????????数据存储在大数据集群不同节点上

数据集：

????????RDD封装了计算逻辑，并不保存数据

数据抽象：

????????RDD是一个抽象类，需要子类具体实现

不可变：

????????RDD封装了计算逻辑，是不可以改变的，想要改变，只能产生新的RDD，在新的RDD里面封装计算逻辑（有向无环图DAG）

可分区：

????????实现并行计算

2.RDD的核心属性

分区列表

????????RDD数据结构中存在分区列表，用于执行任务时并行计算，是实现分布式计算的重要属性

分区计算函数

????????Spark在计算时，是使用分区函数对每一个分区进行计算

RDD之间的依赖关系

????????RDD是计算模型的封装，当需求中需要将多个计算模型进行组合是，就需要将多个RDD建立依赖关系

分区器（可选）

????????当数据为KV类型数据时，可以通过设定分区器自定义数据的分区

首选位置（可选）

????????计算数据时，可以根据计算节点的状态选择不同的节点位置进行计算

3.RDD工作原理

?1）启动Yarn集群环境

?2）Spark通过申请资源创建调度节点和计算节点

?3）Spark框架根据需求将计算逻辑根据分区划分成不同的任务

?4）调度节点将任务根据计算节点状态发送到对应的计算节点进行计算

?从以上流程可以看出RD在整个流程中主要用于讲逻辑进行封装，并生成Task发送给Executor节点执行计算

4.RDD创建

在Spark中创建RDD的方式可以分为四种：

1）从集合（内存）中创建RDD

从集合中创建RDD，Spark主要提供了两个方法：parallelize和makeRDD

    val conf = new SparkConf().setMaster("local[*]").setAppName("test1")
    val sc = new SparkContext(conf)
    val rdd1 = sc.parallelize(List(1, 2, 3, 4))
    val rdd2 = sc.makeRDD(List(1, 2, 3, 4))
    println(rdd1.collect().mkString(","))
    println(rdd2.collect().mkString(","))
    sc.stop()

从底层代码实现来讲，makeRDD方法其实就是parallelize方法

  /** Distribute a local Scala collection to form an RDD.
   *
   * This method is identical to `parallelize`.
   * @param seq Scala collection to distribute
   * @param numSlices number of partitions to divide the collection into
   * @return RDD representing distributed collection
   */
    def makeRDD[T: ClassTag](
      seq: Seq[T],
      numSlices: Int = defaultParallelism): RDD[T] = withScope {
    parallelize(seq, numSlices)
  }

2）从外部存储（文件）创建RDD

由外部存储系统的数据集创建RDD包括：本地的文件系统，所有Hadoop支持的数据集，比如HDFS、HBase等

    val conf = new SparkConf().setMaster("local[*]").setAppName("test1")
    val sc = new SparkContext(conf)
    val rdd = sc.textFile("file:///D:\\Study\\13_spark\\cha01\\file\\tweets.txt")
    rdd.collect().foreach(println)
    sc.stop()

3）从其他RDD创建

主要是通过一个RDD运算完后，在产生新的RDD

4）直接创建RDD（new）

使用new的方式直接构造RDD一般由Spark框架自身使用

5.RDD算子

Value类型

1）map

作用：返回一个新的RDD，该RDD由每一个输入元素经过func函数转换后组成
需求：创建一个RDD，使每个元素*2组成新的RDD

object SparkRdd {
  def main(args: Array[String]): Unit = {
    //1、创建本地spark配置文件
    val config: SparkConf = new SparkConf().setMaster("local[*]").setAppName("WordCount")
    //2、创建Spark上下文对象
    val sc = new SparkContext(config)
    //3、创建RDD
    val listRdd: RDD[Int] = sc.makeRDD(List(1, 2, 3, 4, 5))
    //4、每个数变成原来的2倍
    val rsRdd: RDD[Int] = listRdd.map(_ * 2)
    //5、打印
    rsRdd.foreach(println(_))
  }
}

2）mapPartitions

类似于map，但独立地在RDD的每一个分片上运行，因此在类型为T的RDD上运行时，func的函数类型必须是Iterator[T] => Iterator[U]。假设有N个元素，有M个分区，那么map的函数的将被调用N次,而mapPartitions被调用M次,一个函数一次处理所有分区。
需求：创建一个RDD，使每个元素*2组成新的RDD

object SparkRdd {
  def main(args: Array[String]): Unit = {
    //1、创建本地spark配置文件
    val config: SparkConf = new SparkConf().setMaster("local[*]").setAppName("WordCount")
    //2、创建Spark上下文对象
    val sc = new SparkContext(config)
    //3、创建RDD
    val listRdd: RDD[Int] = sc.makeRDD(List(1, 2, 3, 4, 5))
    //4、每个数变成原来的2倍
    val rsRdd: RDD[Int] = listRdd.mapPartitions(i=>{i.map(_*2)})
    //5、打印
    rsRdd.foreach(println)
  }
}

3）mapPartitionsWithIndex(func)

类似于mapPartitions，但func带有一个整数参数表示分片的索引值，因此在类型为T的RDD上运行时，func的函数类型必须是(Int, Interator[T]) => Iterator[U]；
需求：创建一个RDD，使每个元素跟所在分区形成一个元组组成一个新的RDD

object SparkRdd {
  def main(args: Array[String]): Unit = {
    //1、创建本地spark配置文件
    val config: SparkConf = new SparkConf().setMaster("local[*]").setAppName("WordCount")
    //2、创建Spark上下文对象
    val sc = new SparkContext(config)
    //3、创建2个分区的RDD
    val listRdd: RDD[Int] = sc.makeRDD(List(1, 2, 3, 4, 5), 2)
    //4、每个数变成原来的2倍
    val rsRdd: RDD[(Int, Int)] = listRdd.mapPartitionsWithIndex((index, item) => {
      item.map((index, _))
    })
    //5、打印
    rsRdd.foreach(println)
  }
}

4）flatMap(func)?

对集合中每个元素进行操作然后再扁平化。
需求：把集合中元素按照’ '提取出来

object SparkRdd {
  def main(args: Array[String]): Unit = {
    //1、创建本地spark配置文件
    val config: SparkConf = new SparkConf().setMaster("local[*]").setAppName("WordCount")
    //2、创建Spark上下文对象
    val sc = new SparkContext(config)
    //3、创建2个分区的RDD
     val listRdd: RDD[String] = sc.makeRDD(List("Hello World","Hello Spark"))
    //4、扁平化
    val rsRdd: RDD[String] = listRdd.flatMap(_.split(" "))
    //5、打印
    rsRdd.foreach(println)
  }
}

5）glom

作用：将每一个分区形成一个数组，形成新的RDD类型时RDD[Array[T]]
需求：创建一个3个分区的RDD，并将每个分区的数据放到一个数组

object SparkRdd {
  def main(args: Array[String]): Unit = {
    //1、创建本地spark配置文件
    val config: SparkConf = new SparkConf().setMaster("local[*]").setAppName("WordCount")
    //2、创建Spark上下文对象
    val sc = new SparkContext(config)
    //3、创建3个分区的RDD
    val listRdd: RDD[Int] = sc.makeRDD(List(1, 2, 3, 4, 5, 6, 7, 8), 3)
    //4、glom
    val rsRdd: RDD[Array[Int]] = listRdd.glom()
    //5、打印
    rsRdd.foreach(item => {
      println(item.mkString(","))
    })
  }
}

6）groupBy

分组，按照传入函数的返回值进行分组。将相同的key对应的值放入一个迭代器。
需求：创建一个RDD，按照元素模以2的值进行分组。

def main(args: Array[String]): Unit = {
  //1、创建本地spark配置文件
  val config: SparkConf = new SparkConf().setMaster("local[*]").setAppName("WordCount")
  //2、创建Spark上下文对象
  val sc = new SparkContext(config)
  //3、创建RDD
  val listRdd: RDD[Int] = sc.makeRDD(1 to 20)
  //4、分组
  val rsRdd: RDD[(Int, Iterable[Int])] = listRdd.groupBy(_%2)
  //5、打印
  rsRdd.foreach(t=>println(t._1,t._2))
}

7）filter

作用：过滤。返回一个新的RDD，该RDD由经过func函数计算后返回值为true的输入元素组成。
需求：创建一个RDD ，过滤出和2取余为0的数

object SparkRdd {
  def main(args: Array[String]): Unit = {
    //1、创建本地spark配置文件
    val config: SparkConf = new SparkConf().setMaster("local[*]").setAppName("WordCount")
    //2、创建Spark上下文对象
    val sc = new SparkContext(config)
    //3、创建RDD
    val listRdd: RDD[Int] = sc.makeRDD(1 to 10)
    //4、过滤
    val rsRdd: RDD[Int] = listRdd.filter(_%2==0)
    //5、打印
    rsRdd.foreach(println)
  }
}

8）distinct

去重分区

object SparkRdd {
  def main(args: Array[String]): Unit = {
    //1、创建本地spark配置文件
    val config: SparkConf = new SparkConf().setMaster("local[*]").setAppName("WordCount")
    //2、创建Spark上下文对象
    val sc = new SparkContext(config)
    //3、创建RDD
    val listRdd: RDD[Int] = sc.makeRDD(List(1,2,1,2,1,2))
    //4、用2个分区保存结果
    val rsRdd: RDD[Int] = listRdd.distinct(2)
    //5、打印
    rsRdd.foreach(println)
  }
}

9）repartition(numPartitions)

重分区，根据分区数，重新通过网络随机洗牌所有数据。
需求：创建一个4个分区的RDD，对其重新分区

object SparkRdd {
  def main(args: Array[String]): Unit = {
    //1、创建本地spark配置文件
    val config: SparkConf = new SparkConf().setMaster("local[*]").setAppName("WordCount")
    //2、创建Spark上下文对象
    val sc = new SparkContext(config)
    //3、创建4个分区的Rdd
    val listRdd: RDD[Int] = sc.makeRDD(1 to 10, 4)
    println("repartition before:" + listRdd.partitions.size)
    //4、重新分区
    val rsRdd: RDD[Int] = listRdd.repartition(2)
    //5、打印
    println("repartition after:" + rsRdd.partitions.size)
  }
}

10）sortBy(func,[ascending], [numTasks])

作用；使用func先对数据进行处理，按照处理后的数据比较结果排序，默认为正序。
需求：创建一个RDD，按照不同的规则进行排序

object SparkRdd {
  def main(args: Array[String]): Unit = {
    //1、创建本地spark配置文件
    val config: SparkConf = new SparkConf().setMaster("local[*]").setAppName("WordCount")
    //2、创建Spark上下文对象
    val sc = new SparkContext(config)
    //3、创建4个分区的Rdd
    val listRdd: RDD[Int] = sc.makeRDD(1 to 10)
    //4、升序
    val sortAsc: Array[Int] = listRdd.sortBy(x => x).collect()
    println(sortAsc.mkString(","))
    //5、降序
    val sortDesc: Array[Int] = listRdd.sortBy(x => x, false).collect()
    println(sortDesc.mkString(","))
  }
}

双Value类型交互

1） union(otherDataset)

作用：对源RDD和参数RDD求并集后返回一个新的RDD
需求：创建两个RDD，求并集

object SparkRdd {
  def main(args: Array[String]): Unit = {
    //1、创建本地spark配置文件
    val config: SparkConf = new SparkConf().setMaster("local[*]").setAppName("WordCount")
    //2、创建Spark上下文对象
    val sc = new SparkContext(config)
    //3、创建2个list
    val list1: RDD[Int] = sc.makeRDD(1 to 2)
    val list2: RDD[Int] = sc.makeRDD(2 to 4)
    //4、求并级
    val rsRdd: Array[Int] = list1.union(list2).collect()
    //5、打印
    rsRdd.foreach(println)
  }
}

2）subtract (otherDataset)

差集，计算差的一种函数，去除两个RDD中相同的元素，不同的RDD将保留下来，创建两个RDD，求第一个RDD与第二个RDD的差集

object SparkRdd {
  def main(args: Array[String]): Unit = {
    //1、创建本地spark配置文件
    val config: SparkConf = new SparkConf().setMaster("local[*]").setAppName("WordCount")
    //2、创建Spark上下文对象
    val sc = new SparkContext(config)
    //3、创建2个list
    val list1: RDD[Int] = sc.makeRDD(1 to 10)
    val list2: RDD[Int] = sc.makeRDD(5 to 10)
    //4、求并级
    val rsRdd: Array[Int] = list1.subtract(list2).collect()
    //5、打印
    rsRdd.foreach(println)
  }
}

3）intersection(otherDataset)

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

object SparkRdd {
  def main(args: Array[String]): Unit = {
    //1、创建本地spark配置文件
    val config: SparkConf = new SparkConf().setMaster("local[*]").setAppName("WordCount")
    //2、创建Spark上下文对象
    val sc = new SparkContext(config)
    //3、创建2个list
    val list1: RDD[Int] = sc.makeRDD(1 to 10)
    val list2: RDD[Int] = sc.makeRDD(5 to 10)
    //4、交集
    val rsRdd: Array[Int] = list1.intersection(list2).collect()
    //5、打印
    rsRdd.foreach(println)
  }
}

Key-Value类型

1）groupByKey

groupByKey也是对每个key进行操作，但只生成一个sequence。
需求：wordcount

object SparkRdd {
  def main(args: Array[String]): Unit = {
    //1、创建本地spark配置文件
    val config: SparkConf = new SparkConf().setMaster("local[*]").setAppName("WordCount")
    //2、创建Spark上下文对象
    val sc = new SparkContext(config)
    //3、创建list
    val list: RDD[String] = sc.makeRDD(List("Hello World","Hello Scala","Spark Spark Spark"))
    //4、wordcount
     val mapRdd: RDD[(String, Int)] = list.flatMap(_.split(" ")).map((_,1))
    //5、groupByKey
    val groupByKeyRdd: RDD[(String, Iterable[Int])] = mapRdd.groupByKey()
    groupByKeyRdd.foreach(println)
    //6、聚合
    val wcRdd: RDD[(String, Int)] = groupByKeyRdd.map(t=>(t._1,t._2.size))
    wcRdd.foreach(println)
  }
}

2）reduceByKey

在一个(K,V)的RDD上调用，返回一个(K,V)的RDD，使用指定的reduce函数，将相同key的值聚合到一起，reduce任务的个数可以通过第二个可选的参数来设置。
需求：wordcount

object SparkRdd {
  def main(args: Array[String]): Unit = {
    //1、创建本地spark配置文件
    val config: SparkConf = new SparkConf().setMaster("local[*]").setAppName("WordCount")
    //2、创建Spark上下文对象
    val sc = new SparkContext(config)
    //3、创建list
    val list: RDD[String] = sc.makeRDD(List("Hello World","Hello Scala","Spark Spark Spark"))
    //4、扁平化
     val mapRdd: RDD[(String, Int)] = list.flatMap(_.split(" ")).map((_,1))
    //5、wordcount
    val reduceMapRdd: RDD[(String, Int)] = mapRdd.reduceByKey(_+_)
    //6、打印
    reduceMapRdd.foreach(println)
  }
}

3）sortByKey([ascending], [numTasks])

作用：在一个(K,V)的RDD上调用，K必须实现Ordered接口，返回一个按照key进行排序的(K,V)的RDD
创建一个pairRDD，按照key的正序和倒序进行排序

object SparkRdd {
  def main(args: Array[String]): Unit = {
    //1、创建本地spark配置文件
    val config: SparkConf = new SparkConf().setMaster("local[*]").setAppName("WordCount")
    //2、创建Spark上下文对象
    val sc = new SparkContext(config)
    //3、创建list
    val listRdd: RDD[(Int, Int)] = sc.makeRDD(List((1, 3), (1, 2), (1, 4), (2, 3), (3, 6), (3, 8)))
    //4、升序
    val sortAsc: Array[(Int, Int)] = listRdd.sortByKey(true).collect()
    println("升序:" + sortAsc.mkString(","))
    //5、降序
    val sortDesc: Array[(Int, Int)] = listRdd.sortByKey(false).collect()
    println("降序:" + sortDesc.mkString(","))
  }
}

4）join(otherDataset, [numTasks])

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

object SparkRdd {
  def main(args: Array[String]): Unit = {
    //1、创建本地spark配置文件
    val config: SparkConf = new SparkConf().setMaster("local[*]").setAppName("WordCount")
    //2、创建Spark上下文对象
    val sc = new SparkContext(config)
    val r1: RDD[(Int, String)] = sc.makeRDD(Array((1, "a"), (2, "b"), (3, "c")))
    val r2: RDD[(Int, String)] = sc.makeRDD(Array((1, "a"), (2, "b"), (3, "c")))
    val rsRdd: Array[(Int, (String, String))] = r1.join(r2).collect()
    rsRdd.foreach(println)
  }
}

Action算子

1）reduce，作用：通过func函数聚集RDD中的所有元素，先聚合分区内数据，再聚合分区间数据。
需求：创建一个RDD，将所有元素聚合得到结果。

object SparkRdd {
  def main(args: Array[String]): Unit = {
    //1、创建本地spark配置文件
    val config: SparkConf = new SparkConf().setMaster("local[*]").setAppName("WordCount")
    //2、创建Spark上下文对象
    val sc = new SparkContext(config)
    //3、创建RDD
    val rdd: RDD[Int] = sc.makeRDD(1 to 5)
    //4、reduce
    val rs: Int = rdd.reduce(_+_)
    //5、结果
    println(rs)
  }
}

2） collect()

在驱动程序中，以数组的形式返回数据集的所有元素。
创建一个RDD，并将RDD内容收集到Driver端打印

object SparkRdd {
  def main(args: Array[String]): Unit = {
    //1、创建本地spark配置文件
    val config: SparkConf = new SparkConf().setMaster("local[*]").setAppName("WordCount")
    //2、创建Spark上下文对象
    val sc = new SparkContext(config)
    //3、创建RDD
    val lisrRdd: RDD[Int] = sc.makeRDD(List(1,2,3,4,5))
    val ints: Array[Int] = lisrRdd.collect()
    println(ints.mkString(","))
  }
}

3） count()

返回RDD中元素的个数

object SparkRdd {
  def main(args: Array[String]): Unit = {
    //1、创建本地spark配置文件
    val config: SparkConf = new SparkConf().setMaster("local[*]").setAppName("WordCount")
    //2、创建Spark上下文对象
    val sc = new SparkContext(config)
    //3、创建RDD
     val rdd: RDD[Int] = sc.makeRDD(1 to 10)
    val rddCount: Long = rdd.count()
    println(rddCount)
  }
}

4）first()

返回RDD中的第一个元素

object SparkRdd {
  def main(args: Array[String]): Unit = {
    //1、创建本地spark配置文件
    val config: SparkConf = new SparkConf().setMaster("local[*]").setAppName("WordCount")
    //2、创建Spark上下文对象
    val sc = new SparkContext(config)
    //3、创建RDD
    val rdd: RDD[Int] = sc.makeRDD(1 to 10)
    //4、返回RDD中第一个元素
    val firstNum: Int = rdd.first()
    println(firstNum)
  }
}

5） take(n)

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

object SparkRdd {
  def main(args: Array[String]): Unit = {
    //1、创建本地spark配置文件
    val config: SparkConf = new SparkConf().setMaster("local[*]").setAppName("WordCount")
    //2、创建Spark上下文对象
    val sc = new SparkContext(config)
    //3、创建RDD
    val rdd: RDD[Int] = sc.makeRDD(1 to 10)
    val top3: Array[Int] = rdd.take(3)
    top3.foreach(println)
  }
}

6） takeOrdered(n)

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

def main(args: Array[String]): Unit = {
  //1、创建本地spark配置文件
  val config: SparkConf = new SparkConf().setMaster("local[*]").setAppName("WordCount")
  //2、创建Spark上下文对象
  val sc = new SparkContext(config)
  //3、创建RDD
  val rdd: RDD[Int] = sc.makeRDD(List(5, 3, 2, 1, 7, 6))
  val top3: Array[Int] = rdd.takeOrdered(3)
  println(top3.mkString(","))
}

7）countByKey()

针对(K,V)类型的RDD，返回一个(K,Int)的map，表示每一个key对应的元素个数。

object SparkRdd {
  def main(args: Array[String]): Unit = {
    //1、创建本地spark配置文件
    val config: SparkConf = new SparkConf().setMaster("local[*]").setAppName("WordCount")
    //2、创建Spark上下文对象
    val sc = new SparkContext(config)
    //3、创建RDD
    val rdd: RDD[(Int, Int)] = sc.makeRDD(List((1,3),(1,2),(1,4),(2,3),(3,6),(3,8)))
    //4、统计key
    val countKey: collection.Map[Int, Long] = rdd.countByKey()
    //5、打印
    println(countKey)
  }
}

8）foreach

打印

object SparkRdd {
  def main(args: Array[String]): Unit = {
    //1、创建本地spark配置文件
    val config: SparkConf = new SparkConf().setMaster("local[*]").setAppName("WordCount")
    //2、创建Spark上下文对象
    val sc = new SparkContext(config)
    //3、创建RDD
    val rdd: RDD[Int] = sc.makeRDD(List(1,2,3,4,5))
    rdd.foreach(println)
  }
}