本文主要介绍Spark Core的核心内容&＃xff1a;RDD。包含以下章节和对应的内容
| 章节 | 内容 |
| 1 | RDD简介 |
| 2 | RDD分区 |
| 3 | RDD的依赖关系 |
| 4 | RDD的缓存机制和区别 |
| 5 | RDD创建的两种方式 |
| 6 | RDD算子和总结 |
| 7 | RDD 算子操作案例 |
1、RDD简介
RDD(Resilient Distributed Datasets,弹性分布式数据集)&＃xff0c;是Spark最为核心的概念&＃xff0c;自然也是理解Apache Spark 工作原理的最佳入口之一。
RDD的特点&＃xff1a;
- 是一个分区的只读记录的集合&＃xff1b;
- 一个具有容错机制的特殊集&＃xff1b;
- 只能通过在稳定的存储器或其他RDD上的确定性操作&＃xff08;转换&＃xff09;来创建&＃xff1b;
- 可以分布在集群的节点上&＃xff0c;以函数式操作集合的方式&＃xff0c;进行各种并行操作
RDD之所以为“弹性”的特点
- 基于Lineage的高效容错&＃xff08;第n个节点出错&＃xff0c;会从第n-1个节点恢复&＃xff0c;血统容错&＃xff09;&＃xff1b;
- Task如果失败会自动进行特定次数的重试&＃xff08;默认4次&＃xff09;&＃xff1b;
- Stage如果失败会自动进行特定次数的重试&＃xff08;可以值运行计算失败的阶段&＃xff09;&＃xff0c;只计算失败的数据分片&＃xff1b;
- 数据调度弹性&＃xff1a;DAG TASK 和资源管理无关&＃xff1b;
- checkpoint&＃xff1b;
- 自动的进行内存和磁盘数据存储的切换
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;
- 如何查看分区运行在哪个Worker(机器)上&＃xff1f;
- 如何查看每个分区中的数据&＃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;。
- 窄依赖指的是每一个parent RDD的Partition最多被子RDD的一个Partition使用
- 宽依赖指的是多个子RDD的Partition会依赖同一个parent RDD的Partition
4、RDD的缓存机制和区别
RDD缓存机制有两种&＃xff0c;cache和pesist&＃xff0c;两种区别如下&＃xff1a;
- cache和persist都是用于将一个RDD进行缓存的&＃xff0c;这样在之后使用的过程中就不需要重新计算了&＃xff0c;可以大大节省程序运行时间&＃xff1b;
- cache只有一个默认的缓存级别MEMORY_ONLY &＃xff0c;cache调用了persist&＃xff0c;而persist可以根据情况设置其它的缓存级别&＃xff1b;3&＃xff09;
- 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 算子大致可以分为以下三类:
- Value数据类型的Transformation算子&＃xff0c;这种变换并不触发提交作业&＃xff0c;针对处理的数据项是Value型的数据。
- Key-Value数据类型的Transfromation算子&＃xff0c;这种变换并不触发提交 作业&＃xff0c;针对处理的数据项是Key-Value型的数据对。
- Action算子&＃xff0c;这类算子会触发SparkContext提交Job作业
更多详细算子介绍可以参考&＃xff1a;Scala官方API
以下表格是关于Transformation算子-会延时加载&＃xff08;计算&＃xff09;
| 转换 | 含义 |
| map(func) | 返回一个新的RDD&＃xff0c;该RDD由每一个输入元素经过func函数转换后组成 |
| filter(func) | 返回一个新的RDD&＃xff0c;该RDD由经过func函数计算后返回值为true的输入元素组成 |
| flatMap(func) | 类似于map&＃xff0c;但是每一个输入元素可以被映射为0或多个输出元素&＃xff08;所以func应该返回一个序列&＃xff0c;而不是单一元素&＃xff09; |
| mapPartitions(func) | 类似于map&＃xff0c;但独立地在RDD的每一个分片上运行&＃xff0c;因此在类型为T的RDD上运行时&＃xff0c;func的函数类型必须是Iterator[T] &＃61;> Iterator[U] |
| mapPartitionsWithIndex(func) | 类似于mapPartitions&＃xff0c;但func带有一个整数参数表示分片的索引值&＃xff0c;因此在类型为T的RDD上运行时&＃xff0c;func的函数类型必须是(Int, Interator[T]) &＃61;> Iterator[U] |
| sample(withReplacement, fraction, seed) | 根据fraction指定的比例对数据进行采样&＃xff0c;可以选择是否使用随机数进行替换&＃xff0c;seed用于指定随机数生成器种子 |
| union(otherDataset) | 对源RDD和参数RDD求并集后返回一个新的RDD |
| intersection(otherDataset) | 对源RDD和参数RDD求交集后返回一个新的RDD |
| distinct([numTasks])) | 对源RDD进行去重后返回一个新的RDD |
| groupByKey([numTasks]) | 在一个(K,V)的RDD上调用&＃xff0c;返回一个(K, Iterator[V])的RDD |
| reduceByKey(func, [numTasks]) | 在一个(K,V)的RDD上调用&＃xff0c;返回一个(K,V)的RDD&＃xff0c;使用指定的reduce函数&＃xff0c;将相同key的值聚合到一起&＃xff0c;与groupByKey类似&＃xff0c;reduce任务的个数可以通过第二个可选的参数来设置 |
| aggregateByKey(zeroValue)(seqOp,combOp,[numTasks]) | |
| sortByKey([ascending], [numTasks]) | 在一个(K,V)的RDD上调用&＃xff0c;K必须实现Ordered接口&＃xff0c;返回一个按照key进行排序的(K,V)的RDD |
| sortBy(func,[ascending], [numTasks]) | 与sortByKey类似&＃xff0c;但是更灵活 |
| join(otherDataset, [numTasks]) | 在类型为(K,V)和(K,W)的RDD上调用&＃xff0c;返回一个相同key对应的所有元素对在一起的(K,(V,W))的RDD |
| cogroup(otherDataset, [numTasks]) | 在类型为(K,V)和(K,W)的RDD上调用&＃xff0c;返回一个(K,(Iterable,Iterable))类型的RDD |
| cartesian(otherDataset) | 笛卡尔积 |
| pipe(command, [envVars]) | |
| coalesce(numPartitions) | |
| repartition(numPartitions) | |
| repartitionAndSortWithinPartitions(partitioner) | |
以下表格是关于Action算子- 触发计算
| 动作 | 含义 |
| reduce(func) | 通过func函数聚集RDD中的所有元素&＃xff0c;这个功能必须是课交换且可并联的 |
| collect() | 在驱动程序中&＃xff0c;以数组的形式返回数据集的所有元素 |
| count() | 返回RDD的元素个数 |
| first() | 返回RDD的第一个元素&＃xff08;类似于take(1)&＃xff09; |
| take(n) | 返回一个由数据集的前n个元素组成的数组 |
| takeSample(withReplacement,num, [seed]) | 返回一个数组&＃xff0c;该数组由从数据集中随机采样的num个元素组成&＃xff0c;可以选择是否用随机数替换不足的部分&＃xff0c;seed用于指定随机数生成器种子 |
| takeOrdered(n, [ordering]) | |
| saveAsTextFile(path) | 将数据集的元素以textfile的形式保存到HDFS文件系统或者其他支持的文件系统&＃xff0c;对于每个元素&＃xff0c;Spark将会调用toString方法&＃xff0c;将它装换为文件中的文本 |
| saveAsSequenceFile(path) | 将数据集中的元素以Hadoop sequencefile的格式保存到指定的目录下&＃xff0c;可以使HDFS或者其他Hadoop支持的文件系统。 |
| saveAsObjectFile(path) | |
| countByKey() | 针对(K,V)类型的RDD&＃xff0c;返回一个(K,Int)的map&＃xff0c;表示每一个key对应的元素个数。 |
| foreach(func) | 在数据集的每一个元素上&＃xff0c;运行函数func进行更新。 |
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;并且交换了两次
京公网安备 11010802041100号