RDD 算子分类
发布网友
发布时间:2022-11-09 20:20
我来回答
共1个回答
热心网友
时间:2023-11-22 13:27
摘要: 本文主要介绍Spark算子的作用,以及算子的分类。
转换:Transformation , 行动: Action
RDD算子分类,大致可以分为两类,即:
1. Transformation:转换算子,这类转换并不触发提交作业,完成作业中间过程处理。
2. Action:行动算子,这类算子会触发SparkContext提交Job作业。
下面分别对两类算子进行详细介绍:
一:Transformation:转换算子
1. map:
将原来RDD的每个数据项通过map中的用户自定义函数f映射转变为一个新的元素。源码中map算子相当于初始化一个RDD,新RDD叫做MappedRDD(this,sc.clean(f) )。即:
map是对RDD中的每个元素都执行一个指定的函数来产生一个新的RDD。任何原RDD中的元素在新RDD中都有且只有一个元素与之对应。
上述例子中把原RDD中每个元素都乘以3来产生一个新的RDD。
2. mapPartitions:
mapPartitions函数获取到每个分区的迭代器,在函数中通过这个分区整体的迭代器对整个分区的元素进行操作。内部实现是生成MapPartitionsRDD。
上述例子是通过 函数filter 对分区中所有数据进行过滤。
3. mapValues
针对(key,value)型数据中的Value进行操作,而不对Key进行处理。即:
mapValues顾名思义就是输入函数应用于RDD中Kev-Value的Value,原RDD中的Key保持不变,与新的Value一起组成新的RDD中的元素。因此,该函数只适用于元素为KV对的RDD。
4. mapWith:
mapWith是map的另外一个变种,map只需要一个输入函数,而mapWith有两个输入函数。
eg: 把partition index 乘以10,然后加上2作为新的RDD的元素.(3 是将十个数分为三个区)
5. flatMap:
将原来RDD中的每个元素通过函数f转换为新的元素,并将生成的RDD的每个集合中的元素合并为一个集合,内部创建FlatMappedRDD(this,sc.clean() )。即:
与map类似,区别是原RDD中的元素经map处理后只能生成一个元素,而原RDD中的元素经flatmap处理后可生成多个元素来构建新RDD。
eg:对原RDD中的每个元素x产生y个元素(从1到y,y为元素x的值)。
6. flatMapWith:
flatMapWith与mapWith很类似,都是接收两个函数,一个函数把partitionIndex作为输入,输出是一个新类型A;另外一个函数是以二元组(T,A)作为输入,输出为一个序列,这些序列里面的元素组成了新的RDD。
7. flatMapWithValues:
flatMapValues类似于mapValues,不同的在于flatMapValues应用于元素为KV对的RDD中Value。每个一元素的Value被输入函数映射为一系列的值,然后这些值再与原RDD中的Key组成一系列新的KV对。
上述例子中原RDD中每个元素的值被转换为一个序列(从其当前值到5),比如第一个KV对(1,2), 其值2被转换为2,3,4,5。然后其再与原KV对中Key组成一系列新的KV对(1,2),(1,3),(1,4),(1,5)。
8. rece:
rece将RDD中元素两两传递给输入函数,同时产生一个新的值,新产生的值与RDD中下一个元素再被传递给输入函数直到最后只有一个值为止。
eg:对元素求和。
9. receByKey
顾名思义,receByKey就是对元素为KV对的RDD中Key相同的元素的Value进行rece,因此,Key相同的多个元素的值被rece为一个值,然后与原RDD中的Key组成一个新的KV对。
eg:对Key相同的元素的值求和,因此Key为3的两个元素被转为了(3,10)。
10. cartesian:
对两个RDD内的所有元素进行笛卡尔积操作(耗内存),内部实现返回CartesianRDD。
11. Sample:
sample将RDD这个集合内的元素进行采样,获取所有元素的子集。用户可以设定是否有有放回的抽样,百分比,随机种子,进而决定采样方式。
内部实现: SampledRDD(withReplacement,fraction,seed)。
函数参数设置:
� withReplacement=true,表示有放回的抽样。
� withReplacement=false,表示无放回的抽样。
根据fraction指定的比例,对数据进行采样,可以选择是否用随机数进行替换,seed用于指定随机数生成器种子。
12. union:
使用union函数时需要保证两个RDD元素的数据类型相同,返回的RDD数据类型和被合并的RDD元素数据类型相同。并不进行去重操作,保存所有的元素,如果想去重,可以使用distinct()。同时,spark还提供更为简洁的使用union的API,即通过++符号相当于union函数操作。
eg: a 与 b 的联合
去重复:
13. groupBy:
将元素通过函数生成相应的Key,数据就转化为Key-Value格式,之后将Key相同的元素分为一组。
eg:根据数据集中的每个元素的K值对数据分组
14. join:
join对两个需要连接的RDD进行cogroup函数操作,将相同key的数据能偶放到一个分区,在cgroup操作之后形成新RDD对每个key下的元素进行笛卡尔积的操作,返回的结果在展平,对应key下的所有元组形成一个集合。最后返回 RDD[(K, (V, W))]。
eg:a与b两个数据连接,相当于表的关联
15. cache:
cache将RDD元素从磁盘缓存到内存。相当于 persist(MEMORY_ONLY) 函数的
功能。
16. persist:
persist函数对RDD进行缓存操作,数据缓存在哪里,由StorageLevel这个枚举类型进行确定。DISK 代表磁盘,MEMORY 代表内存, SER 代表数据是否进行序列化存储。
函数定义: persist(newLevel:StorageLevel)
StorageLevel 是枚举类型,代表存储模式。
MEMORY_AND_DISK_SER 代表数据可以存储在内存和磁盘,并且以序列化的方式存储,其他同理。
二:Action:行动算子
1. foreach:
foreach对RDD中的每个元素都应用f函数操作,不返回 RDD 和 Array, 而是返回Uint。
2. saveAsTextFile:
函数将数据输出,存储到 HDFS 的指定目录。
函数的内 部实现,其内部通过调用 saveAsHadoopFile 进行实现:
this.map(x => (NullWritable.get(), new Text(x.toString)))
.saveAsHadoopFile[TextOutputFormat[NullWritable, Text]](path)
将 RDD 中的每个元素映射转变为 (null, x.toString),然后再将其写入 HDFS。
3. collect:
collect相当于toArray,不过已经过时不推荐使用,collect将分布式的RDD返回为一个单机的scala Array数据,在这个数组上运用 scala 的函数式操作。
4. count:
count返回整个RDD的元素个数。