欢迎您访问程序员文章站本站旨在为大家提供分享程序员计算机编程知识!
您现在的位置是: 首页  >  IT编程

Spark RDD API详解之:Map和Reduce

程序员文章站 2023-11-18 08:06:28
RDD是什么? RDD是Spark中的抽象数据结构类型,任何数据在Spark中都被表示为RDD。从编程的角度来看, RDD可以简单看成是一个数组。和普通数组的区别是,RDD中的数据是分区存储的,这样不同分区的数据就可以分布在不同的机器上,同时可以被并行处理。因此,spark应用程序所做的无非是把需要 ......

RDD是什么?

RDD是Spark中的抽象数据结构类型,任何数据在Spark中都被表示为RDD。从编程的角度来看,

RDD可以简单看成是一个数组。和普通数组的区别是,RDD中的数据是分区存储的,这样不同分区的数据就可以分布在不同的机器上,同时可以被并行处理。因此,spark应用程序所做的无非是把需要处理的数据转换成RDD,然后对RDD进行一系列的变换和操作从而得到结果。本文为第一部分,将介绍Spark RDD中与Map和Reduce相关的API。

如何创建RDD?

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

方式(一):举例:从普通数组创建RDD,里面包含了1到9这9个数字,他们分别在3个分区中。

scala>val a=sc.parallelize(1 to 9, 3)
a:org.apache.spark.rdd.RDD[Int] = ParallelCollectionRDD[1] at parallelize at <console>:12

方式(二):举例:读取文件README.md来创建RDD,文件中的每一行就是RDD中的一个元素

scala> val b = sc.textFile("README.md")
b: org.apache.spark.rdd.RDD[String] = MappedRDD[3] at textFile at <console>:12

Text file RDDs的创建可以使用SparkContext的textFile方法。该方法接受一个文件的URI地址(或者是机器上的一个本地路径,或者是一个hdfs://,等URI)作为参数,并读取文件的一行数据,放入集合中。下面是一个调用例子:

scala>val distFile = sc.textFile("data.txt")
distFile:RDD[String]= MappedRDD@1D4CEE08

一旦创建完成,就可以在distFile上执行数据集操作。例如:想要对所有行的长度进行求和,我们就可以通过如下的map和reduce操作来完成:

distFile.map(s =>s.length).reduce((a+b) =>a+b)

虽然还有别的方式可以创建RDD,但在本文中我们主要使用上述两种方式来创建RDD以说明RDD的API.

map

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

举例:

Spark RDD API详解之:Map和Reduce
scala>val a = sc.paralleize(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)
Spark RDD API详解之:Map和Reduce

上述例子中把原RDD中每个元素都乘以2来产生一个新的RDD。

mapPartitions

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

它的函数定义为:

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

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

举例:

Spark RDD API详解之:Map和Reduce
scala>val a = sc.parallelize(1 to 9,3)
scala>def myfunc[T](iter:Iterator[T]:Iterator[(T,T)] = {
   var  res = List[(T,T)]()
   var  pre = iter.next.while(iter,hasNext){
    val cur = iter.next;
    res.::=(pre,cur) pre = cur;
} 
   res.iterator


}) 
scala>a.mapPartitions(myfunc).collect
res0:Array[(Int,Int)] = Array((2,3),(1,2),(5,6),(4,5),(8,9),(7.8))
Spark RDD API详解之:Map和Reduce

上述例子中的函数myfunc是把分区中一个元素和它的下一个元素组成一个Tuple。因为分区中最后一个元素没有下一个元素了,所以(3,4)和(6,7)不在结果中。 
mapPartitions还有些变种,比如mapPartitionsWithContext,它能把处理过程中的一些状态信息传递给用户指定的输入函数。还有mapPartitionsWithIndex,它能把分区的index传递给用户指定的输入函数。

mapValuesmapValues顾名思义就是输入函数应用于RDD中key-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,xlions),(3,xcatx),(7,xpantherx),(5,xeaglex))

mapWith

mapWth是map的另外一个变种,map只需要一个输入函数,而mapWth有两个输入函数。它的定义如下: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)

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)

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)

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)。

reduce

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

举例

scala> val c = sc.parallelize(1 to 10)
scala> c.reduce((x, y) => x + y)
res4: Int = 55

上述例子中对RDD中的元素求和。

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)。