RDD 编程（一）

1. RDD 的创建

创建 RDD 有三种方式：

从集合中创建
从外部存储创建
从其他 RDD 转换得到新的 RDD

1.1 从集合中创建

1、使用 parallelize 函数：

import org.apache.spark.sql.SparkSession

object CreateRdd {
  def main(args: Array[String]): Unit = {
    val session = SparkSession.builder.appName("create_rdd").master("local[2]").getOrCreate()
    val sc = session.sparkContext
    val arr = Array(10, 20, 30, 40, 50, 60)
    val rdd1 = sc.parallelize(arr)

    rdd1.collect().foreach(println)

    sc.stop()
  }
}

注意：parallelize 函数第二个参数可以指定分区数目

2、使用 makeRDD 函数：

val rdd1 = sc.makeRDD(arr)

1.2 从外部存储创建

外部存储包括：

任意 Hadoop 支持的存储数据源来创建分布式数据集
本地文件系统, HDFS, Cassandra, HVase, Amazon S3 等等

val rdd1 = sc.textFile(URL)

URL 可以是本地系统文件，但是必须每个节点都要存在这个路径
可以是 hdfs 路径：hdfs://...
所有基于文件的方法，支持目录、压缩文件、通配符（*）:textFile("/my/directory"), textFile("/my/directory/*.txt"), and textFile("/my/directory/*.gz")
textFile 第二个参数，表示分区数，默认情况下每个块对应一个分区（对 HDFS 来说，块大小默认是 128M，HDFS 文件有多少个就有多少个分区），可以传递一个大于块数的分区数，但是不能传递一个比块数小的分区数

1.3 从其他 RDD 转换为新的 RDD

1.4 RDD 操作

RDD 支持两种操作：transformation、action 操作：

转换操作 transformation：从一个已知的 rdd 中创建处理一个新的 rdd，转换操作是惰性的，只有遇到行动操作才会触发计算
行动操作 action：数据集计算结束后，给驱动程序返回一个值

RDD 类型

根据数据类型不同，整体分为：

value 类型
key-value 类型（二维数组）

注意：每次 action 前面的 transformation 都会被重新计算，但可以通过缓存来加快访问速度

2. value 类型

2.1 map(func)

返回一个新的 RDD, 该 RDD 是由原 RDD 的每个元素经过函数转换后的值而组成. 就是对 RDD 中的数据做转换，用来做数据结构的调整

val arr = Array(10, 20, 30, 40, 50, 60)

//    val rdd1 = sc.parallelize(arr).map(x => x * 2)
val rdd1 = sc.parallelize(arr).map(_ * 2)

2.2 mapPartitions(func)

功能类似于 map，但是独立在每个分区上运行，假设有N个元素，有M个分区，那么 map 的函数的将被调用N次，而 mapPartitions 被调用M次，一个函数一次处理所有分区：

// mapPartitions, x.map() 用的是 scala 的 map 函数，而非 spark 的 map
//    val rdd2 = sc.parallelize(arr, 4).mapPartitions(x => x.map(_ * 2))

val rdd2 = sc.parallelize(arr, 4).mapPartitions(x => {
    println("called one times!!!!!!")

    x.map(_ * 3)
})

运行结果：

called one times!!!!!! 会输出 4 次，因为有 4 个分区：

called one times!!!!!!
called one times!!!!!!
called one times!!!!!!
called one times!!!!!!
30
60
90
120
150
180

注意：当每个分区的数据很大时，使用 mapPartitions x.toList() 可能会把内存撑爆，从而导致 OOM；当内存足够大的时候，使用 mapPartitions 执行效率要比 map 高

2.3 mapPartitionsWithIndex(func)

和 mapPartitions(func)类似. 但是会给 func 多提供一个 Int 值来表示分区的索引. 所以func的类型是:(Int, Iterator<T>) => Iterator<U>：

val rdd3 = sc.parallelize(arr, 2).mapPartitionsWithIndex((index, it) => it.map(x => (index, x * 2)))
rdd3.collect().foreach(println)
sc.stop()

运行结果：

(0,20)
(0,40)
(0,60)
(1,80)
(1,100)
(1,120)

2.4 flatMap(func)

flatMap 返回一个序列，输入一个可能会返回 0 个或多个，map 是一一对应的：

// list1 中每个元素都是一个集合，flatMap 会将集合中的元素拍散，放在一个新的集合中
val list1 = List(1 to 5, 6 to 11, 12 to 18, 18 to 25)
val rdd1 = sc.parallelize(list1, 2)

val rdd2 = rdd1.flatMap(x => x)

运行结果：

其他用法：

val list2 = List(30, 5, 70, 6, 1, 20)
val rdd3 = sc.parallelize(list2).flatMap(x => List(x, x * x, x * x * x))

// 结果：2、4、8，else 返回空列表
val list3 = List(1, 2, 3)
val rdd4 = sc.parallelize(list3, 2).flatMap(x => if (x % 2 == 0) List(x, x * x, x * x * x) else List[Int]())

2.5 glom()

将每个分区元素合并成一个数组，形成新的 rdd，类型是 RDD[Array[T]]：

val list1 = List(1, 2, 3, 4, 5, 6)
val rdd = sc.parallelize(list1, 2).glom().map(x => x.toList)
//    val rdd = sc.parallelize(list1, 2).glom().map(_.toList)

运行结果：

List(1, 2, 3)
List(4, 5, 6)

2.6 groupBy(func)

按照 func 的返回值进行分组，func 的返回值作为 key，对应值放入一个迭代器中，返回：RDD[(K, Iterable[T])]，但是顺序不能保证（不推荐使用， groupByKey 用的比较多）

按照元素的奇偶性进行分组：

val rdd = sc.parallelize(List(30, 50, 7, 6, 1, 20), 2)
val rdd2 = rdd.groupBy(x => x % 2)
/*
奇数、偶数聚合
rdd2 =
  (0,CompactBuffer(30, 50, 6, 20))
  (1,CompactBuffer(7, 1))
  
  rdd3
  (0,106)
  (1,8)
  */
val rdd3 = rdd2.map {
  case (k, it) => (k, it.sum)
}
rdd3.collect().foreach(println)

2.7 filter(func)

作用: 过滤. 返回一个新的 RDD 是由 func 的返回值为 true 的那些元素组成

// 过滤出大于 20 的元素
val rdd = sc.parallelize(List(30, 50, 7, 6, 1, 20), 2).filter(x => x > 20)

2.8 sample(withReplacement, fraction, seed)

作用：

以指定的随机种子随机抽样出比例为fraction的数据，(抽取到的数量是: size * fraction). 需要注意的是得到的结果并不能保证准确的比例.
withReplacement 表示是抽出的数据是否放回，true为有放回的抽样，false为无放回的抽样. 放回表示数据有可能会被重复抽取到, false 则不可能重复抽取到. 如果是false, 则fraction必须是:[0,1], 是 true 则大于等于0就可以了.
seed 用于指定随机数生成器种子。一般用默认的, 或者传入当前的时间戳

val rdd = sc.parallelize(1 to 26, 2)

//    不放回抽样，比例[0, 1]
//    val sample_rdd = rdd.sample(false, 0.1)

// 放回抽样
val sample_rdd = rdd.sample(true, 1)

2.9 distinct([numTasks]))

对 RDD 中元素执行去重操作. 参数表示任务的数量，默认值和分区数保持一致：

val list1 = List(1, 2, 3, 4, 5, 6, 6, 4, 3)
val rdd = sc.parallelize(list1, 2).distinct()

模板匹配对象去重：

import org.apache.spark.sql.SparkSession

object Distinct {
  def main(args: Array[String]): Unit = {
    val session = SparkSession.builder.appName("create_rdd").master("local[2]").getOrCreate()
    val sc = session.sparkContext

    val list1 = List(1, 2, 3, 4, 5, 6, 6, 4, 3)
//    val rdd = sc.parallelize(list1, 2).distinct()

    val rdd = sc.parallelize(List(User(10, "lisi"), User(20, "zs"), User(10, "ab")), 2).distinct(2)
    rdd.collect().foreach(println)
    sc.stop()

  }
}

case class User(age: Int, name: String) {
  override def hashCode(): Int = this.age

  override def equals(obj: Any): Boolean = obj match {
    case User(age, _) => this.age == age
    case _ => false
  }
}

注意：distinct 有 shuffle 操作

2.10 coalesce(numPartitions)

作用：减少分区到指定数目，用于大数据集过滤后，提高小数据集的执行效率，不常用，常用 reparation：

由原来 5 个分区减少到 2 个分区：

val rdd = sc.parallelize(List(30, 50, 70, 60, 10, 20), 5)
println("分区前：" + rdd.getNumPartitions)

val rdd2 = rdd.coalesce(2)
println("分区后：" + rdd2.getNumPartitions)

rdd.collect().foreach(println)

coalesce 一般用来减少分区，也可以增加分区，增加分区时一定需要 shuffle，减少分区一般不会 shuffle：

// 增加分区，第二个参数为 true，表示增加分区
rdd.coalesce(6, true)

2.11 repartition(numPartitions)

作用: 根据新的分区数, 重新 shuffle 所有的数据, 这个操作总会通过网络，新的分区数相比以前可多，也可以少：

rdd.repartition(6)

coalasce 和 reparation 的区别

coalasce：重新分区可选择是否进行 shuffle，由第二个参数 Boolean = false/true 决定
reparation：实质调用的 coalasce 进行 shuffle
若减少分区，尽量减少 shuffle 使用 coalasce

2.12 sortBy(func,[ascending], [numTasks])

作用：使用 func 先对数据进行处理，而后对结果进行排序，默认为正序

val list1 = List("aaa", "ccc", "bbb", "ddd", "eee")
val rdd1 = sc.parallelize(list1, 2)
val rdd = rdd1.sortBy(x => x)
//    val rdd = rdd1.sortBy(x => x, ascending = false)
//    val rdd = rdd1.sortBy(x => x.length, ascending = false)

2.13 pipe(command, [envVars])

作用：管道，针对每个分区，把 RDD 中的每个数据通过管道传递给shell命令或脚本，返回输出的 RDD。一个分区执行一次这个命令. 如果只有一个分区, 则执行一次命令，shell 脚本必须在 Worker 节点

1、创建脚本 test_pipe.sh：

#!/bin/bash

echo Hello
while read line;do
    echo ">>>>>>" $line
done

# 记得添加权限，否则第二步调用会报错，没有权限 chmod +x test_pipe.sh

2、spark-shell 中调用 test_pipe.sh：

// 一个分区
scala> val rdd1 = sc.parallelize(List(1, 2, 3, 4, 5, 6))
rdd1: org.apache.spark.rdd.RDD[Int] = ParallelCollectionRDD[0] at parallelize at <console>:24

scala> rdd1.pipe("/home/hadoop/apps/test_pipe.sh").collect()
res5: Array[String] = Array(Hello, >>>>>> 1, >>>>>> 2, >>>>>> 3, >>>>>> 4, >>>>>> 5, >>>>>> 6)

// 二个分区 每个分区执行一次脚本, 但是每个元素算是标准输入中的一行
scala> val rdd2 = sc.parallelize(List(1, 2, 3, 4, 5, 6), 2)
rdd2: org.apache.spark.rdd.RDD[Int] = ParallelCollectionRDD[4] at parallelize at <console>:24

scala> rdd2.pipe("/home/hadoop/apps/test_pipe.sh").collect()
res6: Array[String] = Array(Hello, >>>>>> 1, >>>>>> 2, >>>>>> 3, Hello, >>>>>> 4, >>>>>> 5, >>>>>> 6)

3. 双 Value 类型

所谓双 value 类型即两个 rdd 进行交互，常用的有：union、subtract、zip 等

3.1 union(otherDataset) 并集

union：并集
intersection：交集
subtract：差集
cartesian：笛卡尔积

作用：求并集，返回一个新的 rdd：

val list1 = 1 to 15
val list2 = 13 to 26

val rdd1 = sc.parallelize(list1, 2)
val rdd2 = sc.parallelize(list2, 2)

// 并集
val union_rdd = rdd1.union(rdd2)

// 交集
val intersection_rdd = rdd1.intersection(rdd2)

// 差集
val subtract_rdd = rdd1.subtract(rdd2)

// 笛卡尔积
val cartesian_rdd = rdd1.cartesian(rdd2)

3.2 zip(otherDataset) 并集

作用：拉链操作，两个 rddd 的元素和分区数必须一致，否则异常（scala 中元素个数可以不一致）

1、元素个数、分区数一致：

val list1 = 1 to 5
val list2 = List("A", "B", "C", "D", "E")
val rdd1 = sc.parallelize(list1, 2)
val rdd2 = sc.parallelize(list2, 2)
val rdd = rdd1.zip(rdd2)

运行结果：

(1,A)
(2,B)
(3,C)
(4,D)
(5,E)

2、分区数不一致：

val list1 = 1 to 5
val list2 = List("A", "B", "C", "D", "E")
val rdd1 = sc.parallelize(list1, 2)
val rdd2 = sc.parallelize(list2, 3)
val rdd = rdd1.zip(rdd2)

运行结果：

Can't zip RDDs with unequal numbers of partitions: List(2, 3)

3、其他函数：

// 和元素索引组成一个元组 （元素，索引），如：(A,0)、(B,1)
//    val rdd = rdd2.zipWithIndex()

// 将多个RDD按照partition组合成为新的RDD，该函数需要组合的RDD具有相同的分区数，但对于每个分区内的元素数量没有要求
val rdd = rdd1.zipPartitions(rdd2)((it1, it2) => {
  it1.zipAll(it2, 100, 200)
})

运行结果：

(1,A)
(2,B)
(3,C)
(4,D)
(5,E)

4. key-value（PairRDD）类型

key-value 键值对类型，大多数有 shuffle 操作，其结构类似于：(key, value)，常用的有：groupByKey、reduceByKey、partitionBy 等。

作用: 对 pairRDD 进行分区操作，如果原有的 partionRDD 的分区器和传入的分区器相同, 则返回原 pairRDD，否则会生成 ShuffleRDD，即会产生 shuffle 过程

RDD 编程（一）

1. RDD 的创建

1.1 从集合中创建

1.2 从外部存储创建

1.3 从其他 RDD 转换为新的 RDD

1.4 RDD 操作

2. value 类型

2.1 map(func)

2.2 mapPartitions(func)

2.3 mapPartitionsWithIndex(func)

2.4 flatMap(func)

2.5 glom()

2.6 groupBy(func)

2.7 filter(func)

2.8 sample(withReplacement, fraction, seed)

2.9 distinct([numTasks]))

2.10 coalesce(numPartitions)

2.11 repartition(numPartitions)

2.12 sortBy(func,[ascending], [numTasks])

2.13 pipe(command, [envVars])

3. 双 Value 类型

3.1 union(otherDataset) 并集

3.2 zip(otherDataset) 并集

4. key-value（PairRDD） 类型

4. key-value（PairRDD）类型