Spark：RDD简介及基础算子

本文主要介绍Spark Core的核心内容&＃xff1a;RDD。包含以下章节和对应的内容

1、RDD简介

RDD(Resilient Distributed Datasets,弹性分布式数据集)&＃xff0c;是Spark最为核心的概念&＃xff0c;自然也是理解Apache Spark 工作原理的最佳入口之一。

RDD的特点&＃xff1a;

1. 是一个分区的只读记录的集合&＃xff1b;
2. 一个具有容错机制的特殊集&＃xff1b;
3. 只能通过在稳定的存储器或其他RDD上的确定性操作&＃xff08;转换&＃xff09;来创建&＃xff1b;
4. 可以分布在集群的节点上&＃xff0c;以函数式操作集合的方式&＃xff0c;进行各种并行操作

RDD之所以为“弹性”的特点

1. 基于Lineage的高效容错&＃xff08;第n个节点出错&＃xff0c;会从第n-1个节点恢复&＃xff0c;血统容错&＃xff09;&＃xff1b;
2. Task如果失败会自动进行特定次数的重试&＃xff08;默认4次&＃xff09;&＃xff1b;
3. Stage如果失败会自动进行特定次数的重试&＃xff08;可以值运行计算失败的阶段&＃xff09;&＃xff0c;只计算失败的数据分片&＃xff1b;
4. 数据调度弹性&＃xff1a;DAG TASK 和资源管理无关&＃xff1b;
5. checkpoint&＃xff1b;
6. 自动的进行内存和磁盘数据存储的切换

2、RDD的分区

首先&＃xff0c;RDD是一个逻辑概念&＃xff0c;分区是一个物理概念&＃xff0c;并且带有下标

举例&＃xff1a;通过SparkContext.parallelize创建一个RDD&＃xff1a; 

val rdd1 &＃61; sc.parallelize(Array(1,2,3,4,5,6,7,8),3)      # 后面的数字3 表示3个分区

也就是说&＃xff0c;RDD 内部的数据集合在逻辑上&＃xff08;以及物理上&＃xff09;被划分成多个小集合&＃xff0c;这样的每一个小集合被称为分区。

那么引出两个问题&＃xff1a;

1. 如何查看分区运行在哪个Worker(机器)上&＃xff1f;
2. 如何查看每个分区中的数据&＃xff1f;

针对问题1和2&＃xff1a; 在源码级别&＃xff0c;RDD 类内存储一个 Partition 列表。每个 Partition 对象都包含一个 index 成员&＃xff0c;通过 RDD 编号 &＃43; index 就能从唯一确定分区的 Block 编号&＃xff0c;持久化的 RDD 就能通过这个 Block 编号从存储介质中获得对应的分区数据。

3、RDD的依赖关系

RDD和它依赖的parent RDD(s)的关系有两种不同的类型&＃xff0c;即窄依赖&＃xff08;narrow dependency&＃xff09;和宽依赖&＃xff08;wide dependency&＃xff09;。

1. 窄依赖指的是每一个parent RDD的Partition最多被子RDD的一个Partition使用
2. 宽依赖指的是多个子RDD的Partition会依赖同一个parent RDD的Partition

4、RDD的缓存机制和区别

RDD缓存机制有两种&＃xff0c;cache和pesist&＃xff0c;两种区别如下&＃xff1a;

1. cache和persist都是用于将一个RDD进行缓存的&＃xff0c;这样在之后使用的过程中就不需要重新计算了&＃xff0c;可以大大节省程序运行时间&＃xff1b;
2.  cache只有一个默认的缓存级别MEMORY_ONLY &＃xff0c;cache调用了persist&＃xff0c;而persist可以根据情况设置其它的缓存级别&＃xff1b;3&＃xff09;
3. executor执行的时候&＃xff0c;默认60%做cache&＃xff0c;40%做task操作&＃xff0c;persist最根本的函数&＃xff0c;最底层的函数

5、RDD创建的两种方式

  1&＃xff1a;通过SparkContext.parallelize创建
val rdd1 &＃61; sc.parallelize(Array(1,2,3,4,5,6,7,8),3)    # 3个分区

  2&＃xff1a;通过读取外部的数据源创建&＃xff1a;比如&＃xff1a;HDFS、本地目录

val rdd1 &＃61; sc.textFile("hdfs://bigdata:9000/input/data.txt")    # HDFS文件
val rdd2 &＃61; sc.textFile("/root/temp/data.txt")   # 本地文件

6、RDD算子

RDD中的所有转换都是延迟加载的&＃xff0c;也就是说&＃xff0c;它们并不会直接计算结果。相反的&＃xff0c;它们只是记住这些应用到基础数据集&＃xff08;例如一个文件&＃xff09;上的转换动作。只有当发生一个要求返回结果给Driver的动作时&＃xff0c;这些转换才会真正运行。这种设计让Spark更加有效率地运行。

从大方向来说&＃xff1a;RDD算子分为Transformation算子 和Action 算子&＃xff0c;其中Transformation 操作是延迟计算的&＃xff0c;也就是说从一个RDD 转换生成另一个 RDD 的转换操作不是马上执行&＃xff0c;需要等到有 Action 操作的时候才会真正触发运算。Action 算子会触发 SparkContext 提交 Job 作业&＃xff0c;并将数据输出 Spark系统

从小方向来说&＃xff1a;RDD 算子大致可以分为以下三类:

1. Value数据类型的Transformation算子&＃xff0c;这种变换并不触发提交作业&＃xff0c;针对处理的数据项是Value型的数据。
2. Key-Value数据类型的Transfromation算子&＃xff0c;这种变换并不触发提交 作业&＃xff0c;针对处理的数据项是Key-Value型的数据对。
3. Action算子&＃xff0c;这类算子会触发SparkContext提交Job作业

更多详细算子介绍可以参考&＃xff1a;Scala官方API

以下表格是关于Transformation算子-会延时加载&＃xff08;计算&＃xff09;

以下表格是关于Action算子- 触发计算

7、RDD 算子操作案例&＃xff1a;

案例1&＃xff1a;映射和过滤

创建一个RDD&＃xff0c;也可以使用List
val rdd1 &＃61; sc.parallelize(List(5,6,7,8,1,2,3,100,30))每个元素乘以2&＃xff0c;再排序
val rdd2 &＃61; rdd1.map(_*2).sortBy(x&＃61;>x,true) #降序过滤出大于20的元素
val rdd3 &＃61; rdd2.filter(_ > 20)输出结果
rdd3.collect

案例2&＃xff1a;字符串操作和字符计数

val rdd4 &＃61; sc.parallelize(Array("a b c","d b c","a y c"))val rdd5 &＃61; rdd4.flatMap(_.split(" ")) # 切分字符并切平val rdd6 &＃61; rdd5.map((_,1)) # 逐一统计val rdd7 &＃61; rdd6.reduceByKey(_&＃43;_).collect # 计数输出结果

案例3&＃xff1a;集合操作

集合运算、去重
val rdd6 &＃61; sc.parallelize(List(5,6,7,8,1,2,3,100,30))
val rdd7 &＃61; sc.parallelize(List(1,2,3,4,5,6,7,8,9,10))并集&＃xff1a;union&＃xff08;如果是SQL中的集合运算&＃xff0c;对集合是有要求的&＃xff09;
val rdd8 &＃61; rdd6.union(rdd7)
去重
rdd8.distinct.collect交集: intersect&＃xff08;如果是SQL中的集合运算&＃xff0c;对集合是有要求的&＃xff09;
val rdd9 &＃61; rdd6.intersection(rdd7)

案例4&＃xff1a;key-value操作

val rdd1 &＃61; sc.parallelize(List(("tom", 1), ("jerry", 3), ("kitty", 2), ("shuke", 1)))
val rdd2 &＃61; sc.parallelize(List(("jerry", 2), ("tom", 3), ("shuke", 2), ("kitty", 5)))
val rdd3 &＃61; rdd1.union(rdd2)
//按key进行聚合
val rdd4 &＃61; rdd3.reduceByKey(_ &＃43; _)
rdd4.collect
//按value的降序排序
val rdd5 &＃61; rdd4.map(t &＃61;> (t._2, t._1)).sortByKey(false).map(t &＃61;> (t._2, t._1))
rdd5.collect提示&＃xff1a;交换了key-value的位置&＃xff0c;并且交换了两次