大数据-84 Spark 集群 RDD创建 RDD-Transformation操作算子 详解

点一下关注吧！！！非常感谢！！持续更新！！！

目前已经更新到了：

• Hadoop（已更完）
• HDFS（已更完）
• MapReduce（已更完）
• Hive（已更完）
• Flume（已更完）
• Sqoop（已更完）
• Zookeeper（已更完）
• HBase（已更完）
• Redis （已更完）
• Kafka（已更完）
• Spark（正在更新！）

章节内容

上节我们完成了如下的内容：

• RDD的介绍
• RDD的特点、特点介绍
• Spark 编程模型的介绍

RDD 的创建

SparkContext

SparkContext是编写Spark程序用到的第一个类，是Spark的主要入口点，它负责和整个集群的交互。

• 如果把Spark集群当做服务端，那么Driver就是客户端，SparkContext是客户端的核心
• SparkContext是Spark对外的接口，负责向调用者提供Spark的各种功能
• SparkContext用于连接Spark集群、创建RDD、累加器、广播变量

从集合创建RDD

我们在集群的节点上启动 Spark-Shell 进行学习和测试

spark-shell --master local[*]

如果顺利启动，你就可以看到如下的画面：

尝试运行如下的指令，感受一下

Using Scala version 2.12.10 (OpenJDK 64-Bit Server VM, Java 1.8.0_412)
Type in expressions to have them evaluated.
Type :help formore information.

scala> val rdd1 = sc.parallelize(Array(1,2,3,4,5))
rdd1: org.apache.spark.rdd.RDD[Int]= ParallelCollectionRDD[0] at parallelize at <console>:24

scala> rdd2.getNumPartitions
res1: Int =2

scala> rdd2.partitions.length
res2: Int =2

scala> val rdd3 = sc.makeRDD(List(1,2,3,4,5))
rdd3: org.apache.spark.rdd.RDD[Int]= ParallelCollectionRDD[2] at makeRDD at <console>:24

scala> val rdd4 = sc.makeRDD(1 to 100)
rdd4: org.apache.spark.rdd.RDD[Int]= ParallelCollectionRDD[3] at makeRDD at <console>:24

scala> rdd4.getNumPartitions
res3: Int =2

scala>

对应的截图如下：

从文件系统创建RDD

用 textFile() 方法来从文件系统中加载数据创建RDD，方法将文件的URI作为参数：

• 本地文件系统
• 分布式文件系统 HDFS
• Amazon S3的地址

# 本地系统 注意文件要确保存在
val lines = sc.textFile("file:///opt/wzk/1.txt")# 从分布式文件系统加载
val lines = sc.textFile("hdfs://h121.wzk.icu:9000/wcinput/wordcount.txt")

运行结果如下图所示：

从RDD创建RDD

本质是将一个RDD转换为另一个RDD，从 Transformation

Transformation

RDD的操作算子分为两类：

• Transformation，用来对RDD进行转换，这个操作时延迟执行的（或者是Lazy），Transformation，返回一个新的RDD
• Action，用来触发RDD的计算，得到相关计算结果或者将结果保存到外部系统中，Action：返回int、double、集合（不会返回新的RDD）

每一个Transformation操作都会产生新的RDD，供给下一个“转换”使用
转换得到RDD是惰性求值，也就是说，整个转换过程只有记录了转换的轨迹，并不会发生真正的计算，只有遇到Action操作时，才会发生真正的计算，开始从学院关系（lineage）源头开始，进行物理的转换操作。

常见转换算子1

• map(func)：对数据集中的每个元素都用func，然后返回一个新的RDD
• filter(func)：对数据集中的每个元素都是用func，然后返回一个包含使func为true的元素构成RDD
• flatMap(func)：与map类似，每个输入元素被映射为0或多个输出元素
• mapPartitions(func)：和map很像，但是map是将func作用在每个元素上，而mapPartitions是func作用在整个分区上。假设一个RDD有N个元素，M个分区（N >> M），那么map的函数将被调用N次，而mapPartitions中的函数仅被调用M次，一次处理一个分区中的所有元素
• mapPartitionsWithIndex(func)：与mapPartitions类似，多了分区索引值信息

转换算子1测试

map filter

测试如下的代码：

val rdd1 = sc.parallelize(1 to 10)
val rdd2 = rdd1.map(_*2)
val rdd3 = rdd2.filter(_>10)

执行结果如下图：

我们可以查看当前的结果，但是当前的操作都是Transformation的，并没有真正的执行。
我们需要通过 collect 触发执行，拿到最终的结果

rdd2.collect
rdd3.collect

将会触发执行，可以看到结果为：

flatMap

我们从HDFS加载一个文件过来

val rdd4 = sc.textFile("hdfs://h121.wzk.icu:9000/wcinput/wordcount.txt")
rdd4.collect

执行结果如下图：

我们使用“a”作为分隔符，对这段内容进行分割：

rdd4.flatMap(_.split("a")).collect

执行结果如下图：

mapPartitions

val rdd5 = rdd1.mapPartitions(iter => iter.map(_*2))

执行结果如下

对比 map 和 mapPartitions

上面我们用：

• rdd1.map(_*2)
• rdd1.mapPartitions(iter => iter.map(_*2))

那么这两种有什么区别呢？

• map：每次只处理一条数据
• mapPartitions：每次处理一个分区的数据，分区的数据处理完成后，数据才能释放，资源不足时容易OOM
• 当资源充足时，建议使用 mapPartitions，充分提高处理效率

常见转换算子2

• groupBy（func）：按照传入函数的返回值进行分组，将key相同的值放入一个迭代器
• glom（）：将每一个分区形成一个数组，形成新的RDD类型RDD[Array[T]]
• sample（withReplacement,fraction,seed）：采样算子，以指定的随机数种子seed随机抽样出数量为fraction的数据，withReplacenent表示抽出数据是否放回，true则放回，false不放回
• distinct（[numTasks]）：对RDD元素去重后，返回一个新的RDD，可传入numTasks参数改变RDD分区数
• coalesce（numPartitions）：缩减分区数，没有shuffle
• repartition（numPartitions）：增加或减少分区数，有shuffle
• sortBy（func，[ascending], [numTasks]）：使用func对数据进行处理，对处理后的结果进行排序

宽依赖的算子（shuffle）：groupBy，distinct、repartition、sortBy

转换算子2测试

group by

val rdd1 = sc.parallelize(1 to 10)
val group = rdd1.groupBy(_%3)
group.collect

执行的结果如下图：

glom.map

将 RDD 中元素的每10个元素分组

val rdd1 = sc.parallelize(1 to 101)
val rdd2 = rdd1.glom.map(_.sliding(10, 10).toArray)
rdd2.collect

执行结果如下图：

sample

对数据采样，fraction表示采样的百分比

rdd1.sample(true, 0.2, 2).collect
rdd1.sample(false, 0.2, 2).collect
rdd1.sample(true, 0.2).collect

执行结果如下图：

distinct

对数据进行去重，我们生成一些随机数，然后对这些数值进行去重。

val random = scala.util.Random
val arr =(1 to 20).map(x => random.nextInt(10))
val rdd = sc.makeRDD(arr)
rdd.distinct.collect

执行结果如下图：

numSlices

对RDD重分区，我们需要多分一些区出来

val rdd1 = sc.range(1, 1000, numSlices=10)
val rdd2 = rdd1.filter(_%2==0)
rdd2.getNumPartitions

执行结果如下图：

repartition & coalesce

增加或者减少分区

rdd2.getNumPartitions
# repartition 是增加和缩减分区数
val rdd3 = rdd2.repartition(5)# coalesce 是缩减分区数
val rdd4 = rdd2.coalesce(5)

执行结果如下图：

sortBy

rdd.sortBy(x => x).collect
rdd.sortBy(x => x).collect

执行结果如下：

coalesce & repartition

• repartition：增大或者减少分区数，有shuffle
• coalesce：一般用于减少分区数（此时无shuffle）