Spark RDD API(scala)

1、RDD

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

2、RDD创建

RDD可以从普通数组创建出来,也可以从文件系统或者HDFS中的文件创建出来。

1) 从普通数组创建RDD,里面包含了1到9这9个数字,它们分别在3个分区中

scala> val a = sc.parallelize(1 to 9, 3)

2)读取文件README.md来创建RDD,文件中的每一行就是RDD中的一个元素

scala> val b = sc.textFile("README.md")

3、两类操作算子

主要分两类,转换(transformation)和动作(action)。两类函数的主要区别是,transformation接受RDD并返回RDD,而action接受RDD返回非RDD.

transformation操作是延迟计算的,也就是说从一个RDD生成另一个RDD的转换操作不是马上执行,需要等到有action操作的时候才真正触发运算。

action算子会触发spark提交作业job,并将数据输出spark系统。

4、转换算子

更多可以参看 链接

1)map

对RDD中的每个元素都执行一个指定的函数来产生一个新的RDD。任何原RDD中的元素在新RDD中都有且只有一个元素与之对应。

举例:

scala> val a = sc.parallelize(1 to 9, 3)

scala> val b = a.map(x => x*2)

scala> a.collect

res10: Array[Int] = Array(1, 2, 3, 4, 5, 6, 7, 8, 9)

scala> b.collect

res11: Array[Int] = Array(2, 4, 6, 8, 10, 12, 14, 16, 18)

对比分析下,如果换成flatMap,结果如下:

2)flatMap

与map类似,区别是原RDD中的元素经map处理后只能生成一个元素,而原RDD中的元素经flatmap处理后可生成多个元素来构建新RDD。

举例:对原RDD中的每个元素x产生y个元素(从1到y,y为元素x的值)

scala> val a = sc.parallelize(1 to 4, 2)

scala> val b = a.flatMap(x => 1 to x)

scala> b.collect

res12: Array[Int] = Array(1, 1, 2, 1, 2, 3, 1, 2, 3, 4,1,2,3,4)

对比分析下,如果换成map,结果如下:

3)mapPartitions

mapPartitions是map的一个变种。map的输入函数是应用于RDD中每个元素,而mapPartitions的输入函数是应用于每个分区,也就是把每个分区中的内容作为整体来处理的。

它的函数定义为:

def mapPartitions[U: ClassTag](f: Iterator[T] => Iterator[U], preservesPartitioning: Boolean = false): RDD[U]

f即为输入函数,它处理每个分区里面的内容。每个分区中的内容将以Iterator[T]传递给输入函数f,f的输出结果是Iterator[U]。最终的RDD由所有分区经过输入函数处理后的结果合并起来的。

举例:

上述例子中的函数myfunc是把分区中一个元素和它的下一个元素组成一个Tuple。因为分区中最后一个元素没有下一个元素了,所以(3,4)和(6,7)不在结果中。

mapPartitions还有些变种,比如mapPartitionsWithContext,它能把处理过程中的一些状态信息传递给用户指定的输入函数。还有mapPartitionsWithIndex,它能把分区的index传递给用户指定的输入函数。


4)mapWith

是map的另外一个变种,map只需要一个输入函数,而mapWith有两个输入函数。它的定义如下:

def mapWith[A: ClassTag, U: ](constructA: Int => A, preservesPartitioning: Boolean = false)(f: (T, A) => U): RDD[U]

第一个函数constructA是把RDD的partition index(index从0开始)作为输入,输出为新类型A;

第二个函数f是把二元组(T, A)作为输入(其中T为原RDD中的元素,A为第一个函数的输出),输出类型为U。

举例:把partition index 乘以10,然后加上2作为新的RDD的元素。

val x = sc.parallelize(List(1,2,3,4,5,6,7,8,9,10), 3)

x.mapWith(a => a * 10)((a, b) => (b + 2)).collect

res4: Array[Int] = Array(2, 2, 2, 12, 12, 12, 22, 22, 22, 22)

5)flatMapWith

flatMapWith与mapWith很类似,都是接收两个函数,一个函数把partitionIndex作为输入,输出是一个新类型A;另外一个函数是以二元组(T,A)作为输入,输出为一个序列,这些序列里面的元素组成了新的RDD。它的定义如下:

def flatMapWith[A: ClassTag, U: ClassTag](constructA: Int => A, preservesPartitioning: Boolean = false)(f: (T, A) => Seq[U]): RDD[U]

举例:

scala> val a = sc.parallelize(List(1,2,3,4,5,6,7,8,9), 3)

scala> a.flatMapWith(x => x, true)((x, y) => List(y, x)).collect

res58: Array[Int] = Array(0, 1, 0, 2, 0, 3, 1, 4, 1, 5, 1, 6, 2, 7, 2,8, 2, 9)

6)flatMapValues

flatMapValues类似于mapValues,不同的在于flatMapValues应用于元素为KV对的RDD中Value。每个一元素的Value被输入函数映射为一系列的值,然后这些值再与原RDD中的Key组成一系列新的KV对。

举例

scala> val a = sc.parallelize(List((1,2),(3,4),(3,6)))

scala> val b = a.flatMapValues(x=>x.to(5))

scala> b.collect

res3: Array[(Int, Int)] = Array((1,2), (1,3), (1,4), (1,5), (3,4), (3,5))

上述例子中原RDD中每个元素的值被转换为一个序列(从其当前值到5),比如第一个KV对(1,2), 其值2被转换为2,3,4,5。然后其再与原KV对中Key组成一系列新的KV对(1,2),(1,3),(1,4),(1,5)。

7)union


8)cartesian

9)groupBy

10)filter

当需要比较不同类型数据时,参照 :更多API

11)sample

12)Cache

将RDD元素从磁盘缓存到内存

如果数据需要复用,可以通过Cache算子,将数据缓存到内存。

13)persist

14)mapValues

顾名思义就是输入函数应用于RDD中Kev-Value的Value,原RDD中的Key保持不变,与新的Value一起组成新的RDD中的元素。因此,该函数只适用于元素为KV对的RDD。

举例:

scala> val a = sc.parallelize(List("dog", "tiger", "lion", "cat", "panther", " eagle"), 2)

scala> val b = a.map(x => (x.length, x))

scala> b.mapValues("x" + _ + "x").collect

res5: Array[(Int, String)] = Array((3,xdogx), (5,xtigerx), (4,xlionx),(3,xcatx), (7,xpantherx), (5,xeaglex))

15)combineByKey

16)reduceByKey

顾名思义,reduceByKey就是对元素为KV对的RDD中Key相同的元素的Value进行reduce,因此,Key相同的多个元素的值被reduce为一个值,然后与原RDD中的Key组成一个新的KV对。

举例:

scala> val a = sc.parallelize(List((1,2),(3,4),(3,6)))

scala> a.reduceByKey((x,y) => x + y).collect

res7: Array[(Int, Int)] = Array((1,2), (3,10))

上述例子中,对Key相同的元素的值求和,因此Key为3的两个元素被转为了(3,10)。

17)reduce

reduce将RDD中元素两两传递给输入函数,同时产生一个新的值,新产生的值与RDD中下一个元素再被传递给输入函数直到最后只有一个值为止。

举例:对RDD中的元素求和

scala> val c = sc.parallelize(1 to 10)

scala> c.reduce((x, y) => x + y)

res4: Int = 55

18)join

19)zip


20)intersection

5、Action算子

1)foreach

2)saveAsTextFile

3)collect

相当于toArray,将分布式的RDD返回为一个单机的Scala Array.

4)count

原文引自:https://www.jianshu.com/p/bfeed8f7583d?utm_campaign=maleskine&utm_content=note&utm_medium=seo_notes&utm_source=recommendation

原文地址:https://www.cnblogs.com/jinggangshan/p/8116948.html

时间: 2024-10-12 22:48:54

Spark RDD API(scala)的相关文章

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 Operations(2)

处理数据类型为Value型的Transformation算子可以根据RDD变换算子的输入分区与输出分区关系分为以下几种类型. 1)输入分区与输出分区一对一型. 2)输入分区与输出分区多对一型. 3)输入分区与输出分区多对多型. 4)输出分区为输入分区子集型. 5)还有一种特殊的输入与输出分区一对一的算子类型:Cache型.Cache算子对RDD分区进行缓存. 1.输入分区与输出分区一对一型 (1)map 将原来RDD的每个数据项通过map中的用户自定义函数f映射转变为一个新的元素.源码中的map

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

Azure HDInsight 和 Spark 大数据分析(一)

What is HDInsight? Microsoft Azure HDInsight 是基于 Hortonoworks Data Platform (HDP) 的 Hadoop 集群,包括Storm, HBase, Pig, Hive, Sqoop, Oozie, Ambari等(具体的组件请参看最后的附录).Azure HDInsight 支持 Windows的集群部署,也支持 Linux 集群部署.Hortonworks 是我目前所知唯一支持在 Windows 上部署的 Hadoop C

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

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

用产品思维设计API(三)——版本控制,没有你想的这么简单

用产品思维设计API(三)--版本控制,没有你想的这么简单 前言 最近公司内部在重构项目代码,包括API方向的重构,期间遇到了很多的问题,不由得让我重新思考了下. - 一个优雅的API该如何设计? - 前后端分离之后,API真的解耦分离了吗? - 不断的版本迭代,API的兼容性该如何做? ps.这里所说的API仅为Web API,提供APP\WEB开发使用. 年前,我司内部的接口已经进入了一个完全的重构阶段,参考了市面上各大平台的API和文档,自己也总结出了很多的心得.这里向大家分享一下,接下来

我这么玩Web Api(二):数据验证,全局数据验证与单元测试

目录 一.模型状态 - ModelState 二.数据注解 - Data Annotations 三.自定义数据注解 四.全局数据验证 五.单元测试   一.模型状态 - ModelState 我理解的ModelState是微软在ASP.NET MVC中提出的一种新机制,它主要实现以下几个功能: 1. 保存客户端传过来的数据,如果验证不通过,把数据返回到客户端,这样可以保存用户输入,不需要重新输入. 2. 验证数据,以及保存数据对应的错误信息. 3. 微软的一种DRY(Don't Repeat

【高德地图API】从零开始学高德JS API(七)——定位方式大揭秘

摘要:关于定位,分为GPS定位和网络定位2种.GPS定位,精度较高,可达到10米,但室内不可用,且超级费电.网络定位,分为wifi定位和基站定位,都是通过获取wifi或者基站信息,然后查询对应的wifi或者基站位置数据库,得到的定位地点.定位数据库可以不断完善不断补充,所以,越定位越准确.本文详细描述了,如果使用高德JS API来实现位置定位.城市定位的方法,包含了IP定位,浏览器定位,检索定位等多种网络定位方法.当然,如果您的手机有GPS功能,那么使用浏览器定位的时候,会自动获取GPS信息,使