Spark RDD Operations(2)

处理数据类型为Value型的Transformation算子可以根据RDD变换算子的输入分区与输出分区关系分为以下几种类型。

1)输入分区与输出分区一对一型。

2)输入分区与输出分区多对一型。

3)输入分区与输出分区多对多型。

4)输出分区为输入分区子集型。

5)还有一种特殊的输入与输出分区一对一的算子类型:Cache型。Cache算子对RDD分区进行缓存。

1.输入分区与输出分区一对一型

(1)map

将原来RDD的每个数据项通过map中的用户自定义函数f映射转变为一个新的元素。源码中的map算子相当于初始化一个RDD,新RDD叫作MappedRDD(this, sc.clean(f))。

图3-4中的每个方框表示一个RDD分区,左侧的分区经过用户自定义函数f:T->U映射为右侧的新的RDD分区。但是实际只有等到Action算子触发后,这个f函数才会和其他函数在一个Stage中对数据进行运算。V1输入f转换输出V’1。

(2)flatMap

将原来RDD中的每个元素通过函数f转换为新的元素,并将生成的RDD的每个集合中的元素合并为一个集合。内部创建 FlatMappedRDD(this, sc.clean(f))。

图3-5中小方框表示RDD的一个分区,对分区进行flatMap函数操作,flatMap中传入的函数为f:T->U,T和U可以是任意的数据类型。将分区中的数据通过用户自定义函数f转换为新的数据。外部大方框可以认为是一个RDD分区,小方框代表一个集合。V1、V2、V3在一个集合作为RDD的一个数据项,转换为V’1、V’2、V’3后,将结合拆散,形成为RDD中的数据项。

(3)mapPartitions

mapPartitions函数获取到每个分区的迭代器,在函数中通过这个分区整体的迭代器对整个分区的元素进行操作。内部实现是生成MapPartitionsRDD。图3-6中的方框代表一个RDD分区。

图3-6中,用户通过函数f (iter )=>iter.filter(_>=3)对分区中的所有数据进行过滤,>=3的数据保留。一个方块代表一个RDD分区,含有1、2、3的分区过滤只剩下元素3。

(4)glom

glom函数将每个分区形成一个数组,内部实现是返回的GlommedRDD。图3-7中的每个方框代表一个RDD分区。
图3-7中的方框代表一个分区。该图表示含有V1、V2、V3的分区通过函数glom形成一个数组Array[(V1),(V2),(V3)]。

2.输入分区与输出分区多对一型

(1)union

使用union函数时需要保证两个RDD元素的数据类型相同,返回的RDD数据类型和被合并的RDD元素数据类型相同,并不进行去重操作,保存所有元素。如果想去重,可以使用distinct()。++符号相当于uion函数操作。

图3-8中左侧的大方框代表两个RDD,大方框内的小方框代表RDD的分区。右侧大方框代表合并后的RDD,大方框内的小方框代表分区。含有V1,V2…U4的RDD和含有V1,V8…U8的RDD合并所有元素形成一个RDD。V1、V1、V2、V8形成一个分区,其他元素同理进行合并。

(2)cartesian

对两个RDD内的所有元素进行笛卡尔积操作。操作后,内部实现返回CartesianRDD。图3-9中左侧的大方框代表两个RDD,大方框内的小方框代表RDD的分区。右侧大方框代表合并后的RDD,大方框内的小方框代表分区。

图3-9中的大方框代表RDD,大方框中的小方框代表RDD分区。例如,V1和另一个RDD中的W1、W2、Q5进行笛卡尔积运算形成(V1,W1)、(V1,W2)、(V1,Q5)。

3.输入分区与输出分区多对多型

groupBy:将元素通过函数生成相应的Key,数据就转化为Key-Value 格式,之后将Key相同的元素分为一组。
函数实现如下。

①sc.clean( )函数将用户函数预处理:
val cleanF = sc.clean(f)

②对数据map进行函数操作,最后再对groupByKey进行分组操作。

this.map(t => (cleanF(t), t)).groupByKey(p)

其中,p中确定了分区个数和分区函数,也就决定了并行化的程度。图3-10中的方框代表RDD分区。

图3-10中的方框代表一个RDD分区,相同key的元素合并到一个组。例如,V1,V2合并为一个Key-Value对,其中key为“V”,Value为“V1,V2”,形成V,Seq(V1,V2)。

4.输出分区为输入分区子集型

(1)filter

filter的功能是对元素进行过滤,对每个元素应用f函数,返回值为true的元素在RDD中保留,返回为false的将过滤掉。内部实现相当于生成FilteredRDD(this,sc.clean(f))。

下面代码为函数的本质实现。

def filter(f:T=>Boolean):RDD[T]=new FilteredRDD(this,sc.clean(f))

图3-11中的每个方框代表一个RDD分区。T可以是任意的类型。通过用户自定义的过滤函数f,对每个数据项进行操作,将满足条件,返回结果为true的数据项保留。例如,过滤掉V2、V3保留了V1,将区分命名为V1‘。

(2)distinct

distinct将RDD中的元素进行去重操作。图3-12中的方框代表RDD分区。

图3-12中的每个方框代表一个分区,通过distinct函数,将数据去重。例如,重复数据V1、V1去重后只保留一份V1。

(3)subtract

subtract相当于进行集合的差操作,RDD 1去除RDD 1和RDD 2交集中的所有元素。

图3-13中左侧的大方框代表两个RDD,大方框内的小方框代表RDD的分区。右侧大方框代表合并后的RDD,大方框内的小方框代表分区。V1在两个RDD中均有,根据差集运算规则,新RDD不保留,V2在第一个RDD有,第二个RDD没有,则在新RDD元素中包含V2。

(4)sample

sample将RDD这个集合内的元素进行采样,获取所有元素的子集。用户可以设定是否有放回的抽样、百分比、随机种子,进而决定采样方式。

内部实现是生成SampledRDD(withReplacement, fraction, seed)。

函数参数设置如下。

withReplacement=true,表示有放回的抽样;

withReplacement=false,表示无放回的抽样。

图3-14中的每个方框是一个RDD分区。通过sample函数,采样50%的数据。V1、V2、U1、U2、U3、U4采样出数据V1和U1、U2,形成新的RDD。

(5)takeSample

takeSample()函数和上面的sample函数是一个原理,但是不使用相对比例采样,而是按设定的采样个数进行采样,同时返回结果不再是RDD,而是相当于对采样后的数据进行Collect(),返回结果的集合为单机的数组。

图3-15中左侧的方框代表分布式的各个节点上的分区,右侧方框代表单机上返回的结果数组。通过takeSample对数据采样,设置为采样一份数据,返回结果为V1。

5.Cache型

(1)cache
cache将RDD元素从磁盘缓存到内存,相当于persist(MEMORY_ONLY)函数的功能。图3-14中的方框代表RDD分区。

图3-16中的每个方框代表一个RDD分区,左侧相当于数据分区都存储在磁盘,通过cache算子将数据缓存在内存。

(2)persist

persist函数对RDD进行缓存操作。数据缓存在哪里由StorageLevel枚举类型确定。有以下几种类型的组合(见图3-15),DISK代表磁盘,MEMORY代表内存,SER代表数据是否进行序列化存储。

下面为函数定义,StorageLevel是枚举类型,代表存储模式,用户可以通过图3-17按需选择。

persist(newLevel: StorageLevel)

图3-17中列出persist函数可以缓存的模式。例如,MEMORY_AND_DISK_SER代表数据可以存储在内存和磁盘,并且以序列化的方式存储。其他同理。

图3-18中的方框代表RDD分区。disk代表存储在磁盘,mem代表存储在内存。数据最初全部存储在磁盘,通过persist(MEMORY_AND_DISK)将数据缓存到内存,但是有的分区无法容纳在内存,例如:图3-18中将含有V1,V2,V3的RDD存储到磁盘,将含有U1,U2的RDD仍旧存储在内存。

时间: 2024-10-07 06:34:53

Spark RDD Operations(2)的相关文章

Spark RDD编程(二)

转载请注明出处:http://blog.csdn.net/gamer_gyt @高阳团 博主微博:http://weibo.com/234654758 Github:https://github.com/thinkgamer ============================================================ SparkRDD编程(一) Spark 的键值对(pair RDD)操作,Scala实现 RDD的分区函数 目前Spark中实现的分区函数包括两种 Ha

Spark RDD API(scala)

1.RDD RDD(Resilient Distributed Dataset弹性分布式数据集)是Spark中抽象的数据结构类型,任何数据在Spark中都被表示为RDD.从编程的角度来看,RDD可以简单看成是一个数组.和普通数组的区别是,RDD中的数据时分区存储的,这样不同分区的数据就可以分布在不同的机器上,同时可以被并行处理.因此,Spark应用程序所做的无非是把需要处理的数据转换为RDD,然后对RDD进行一系列的变换和操作,从而得到结果. 2.RDD创建 RDD可以从普通数组创建出来,也可以

Spark RDD初探(一)

本文概要 本文主要从以下几点阐述RDD,了解RDD 什么是RDD? 两种RDD创建方式 向给spark传递函数Passing Functions to Spark 两种操作之转换Transformations 两种操作之行动Actions 惰性求值 RDD持久化Persistence 理解闭包Understanding closures 共享变量Shared Variables 总结 Working with Key-Value Pairs.Shuffle operations.patition

Learning Spark中文版--第三章--RDD编程(1)

? ?本章介绍了Spark用于数据处理的核心抽象概念,具有弹性的分布式数据集(RDD).一个RDD仅仅是一个分布式的元素集合.在Spark中,所有工作都表示为创建新的RDDs.转换现有的RDDs,或者调用RDDs上的操作来计算结果.在底层,Spark自动将数据中包含的数据分发到你的集群中,并将你对它们执行的操作进行并行化.数据科学家和工程师都应该阅读这一章,因为RDDs是Spark的核心概念.我们强烈建议你在这些例子中尝试一些 交互式shell(参见"Spark的Python和Scala she

Spark教程-构建Spark集群(1)

对于90%以上想学习Spark的人而言,如何构建Spark集群是其最大的难点之一,为了解决大家构建Spark集群的一切困难,家林把Spark集群的构建分为了四个步骤,从零起步,不需要任何前置知识,涵盖操作的每一个细节,构建完整的Spark集群. 从零起步,构建Spark集群经典四部曲: 第一步:搭建Hadoop单机和伪分布式环境: 第二步:构造分布式Hadoop集群: 第三步:构造分布式的Spark集群: 第四步:测试Spark集群: 本文内容为构建Spark集群经典四部曲的第一步,从零起步构建

Spark入门实战系列--3.Spark编程模型(下)--IDEA搭建及实战

[注]该系列文章以及使用到安装包/测试数据 可以在<倾情大奉送–Spark入门实战系列>获取 1 安装IntelliJ IDEA IDEA 全称 IntelliJ IDEA,是java语言开发的集成环境,IntelliJ在业界被公认为最好的java开发工具之一,尤其在智能代码助手.代码自动提示.重构.J2EE支持.Ant.JUnit.CVS整合.代码审查. 创新的GUI设计等方面的功能可以说是超常的.IDEA是JetBrains公司的产品,这家公司总部位于捷克共和国的首都布拉格,开发人员以严谨

Spark笔记:RDD基本操作(上)

本文主要是讲解spark里RDD的基础操作.RDD是spark特有的数据模型,谈到RDD就会提到什么弹性分布式数据集,什么有向无环图,本文暂时不去展开这些高深概念,在阅读本文时候,大家可以就把RDD当作一个数组,这样的理解对我们学习RDD的API是非常有帮助的.本文所有示例代码都是使用scala语言编写的. Spark里的计算都是操作RDD进行,那么学习RDD的第一个问题就是如何构建RDD,构建RDD从数据来源角度分为两类:第一类是从内存里直接读取数据,第二类就是从文件系统里读取,当然这里的文件

Spark笔记整理(五):Spark RDD持久化、广播变量和累加器

[TOC] Spark RDD持久化 RDD持久化工作原理 Spark非常重要的一个功能特性就是可以将RDD持久化在内存中.当对RDD执行持久化操作时,每个节点都会将自己操作的RDD的partition持久化到内存中,并且在之后对该RDD的反复使用中,直接使用内存缓存的partition.这样的话,对于针对一个RDD反复执行多个操作的场景,就只要对RDD计算一次即可,后面直接使用该RDD,而不需要反复计算多次该RDD. 巧妙使用RDD持久化,甚至在某些场景下,可以将spark应用程序的性能提升1

spark调优(三):RDD重构和持久化

第一,RDD架构重构与优化 尽量去复用RDD,差不多的RDD,可以抽取称为一个共同的RDD,供后面的RDD计算时,反复使用. 第二,公共RDD一定要实现持久化 对于要多次计算和使用的公共RDD,一定要进行持久化. 持久化,也就是说,将RDD的数据缓存到内存中/磁盘中,(BlockManager),以后无论对这个RDD做多少次计算,那么都是直接取这个RDD的持久化的数据,比如从内存中或者磁盘中,直接提取一份数据. 第三,持久化,是可以进行序列化的 如果正常将数据持久化在内存中,那么可能会导致内存的