Spark 基础操作

Spark 操作

创建操作(Creation Operation)

用于RDD创建工作。RDD创建只有两种方法，一种是来自于内存集合和外部存储系统，另一种是通过转换操作生成的RDD

转换操作(Transformation Operation)

将RDD通过一定的操作变成新的RDD，比如HadoopRDD可以使用map操作变换为MappedRDD

RDD的转换操作是惰性操作，它只是定义了一个新的RDDs,并没有立即执行

控制操作(Control Operation)

进行RDD持久化，可以让RDD按不同的存储策略保存在磁盘或内存中，比如cache接口默认将RDD缓存在内存中

行动操作(Action Operation)

能够触发Spark运行的操作，例如，对RDD进行collect就是行动操作

Spark中行动操作分为两类，一类的操作结果变成Scala集合或者变量，另一类将RDD保存到外部文件系统或者数据库中

创建操作

parallelize[T](seq:Seq[T], numSlices:Int)

parallelize 在一个已经存在的Scala集合上创建(一个Seq对象)。集合的对象将会被复制，创建出一个可以被并行操作的分布式数据集

makeRDD[T](seq:Seq[(T, Seq[String])]):RDD[T]

并行化集合创建操作

基础转换操作

map[U](f:(T) => U):RDD[U]

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

distinct():RDD[(T)]/distinct(nPartitions):RDD[(T)]

distinct操作是去除RDD重复的元素，返回所有元素不重复RDD

flatMap[U](f:(T) => TraversableOnce[U]):RDD[U]

flatMap操作与map类似，区别是原RDD中的每个元素经过map处理后只能生成一个元素，而在flatMap操作中原RDD中的每个元素可生成一个或多个元素来构建RDD

coalesce(numPartitions:Int,shuffle:Boolean = false):RDD[T]

coalesce操作使用HashPartitioner进行重分区，第一个参数为重分区的数目，第二个是否进行shuffle，默认情况为false

repartition(numPartitions:Int):RDD[T]

repartition操作是coalesce函数第二个参数为true的实现

randomSplit(weights:Array[Double], send:Long):Array[RDD[T]]

randomSplit操作是根据weights权重将一个RDD分隔为多个RDD

glom():RDD[Array[T]]

glom操作则是RDD中每一个分区所有类型为T的数据转变成元素为T的数组[Array[T]]

union(other:RDD[T]):RDD[T]

union操作是将两个RDD合并，返回两个RDD的并集，返回元素不去重

intersection(other:RDD[T]):RDD[T]

intersection操作类似SQL中的inner join操作，返回两个RDD交集，返回元素去重

subtract(other:RDD[T]):RDD[T]

subtract 返回在RDD中出现，且不在otherRDD中出现的元素

mapPartitions[U](f:(Iter), preserversPartitons:Boolean)

mapPartitions操作和map操作类似，只不过映射的参数由RDD中的每一个元素变成了RDD中每一个分区的迭代器

其中preserversPartitons表示是否保留父RDD的partitioner分区信息

如果在映射的过程中需要频繁创建额外的对象，使用mapPartitions操作要比map操作高效得多

比如，将RDD中的所有数据通过JDBC连接写入数据库，如果使用map函数，可能要为每一个元素都创建一个connection，这样开销很大

如果使用mapPartitions，那么只需要针对每一个分区建立一个connection

mapPartitionsWithIndex[U](f:(Iter), preserversPartitons:Boolean):RDD[U]

mapPartitionsWithIndex 操作作用类似于mapPartitions，只是输入参数多一个了分区索引

zip[U](other:RDD[U]):RDD[(T,U)]

zip操作用于将两个RDD组合成Key/Value形式的RDD，这里默认两个RDD的partition数量以及元素数量相同，否则会抛出异常

zipPartitions[B,V](rdd2:RDD[B])

zipPartitions 操作将多个RDD按照partition组合成新的RDD，该操作需要组合的RDD具有相同的分区数，但对于每个分区内的元素数量没有要求

zipWithIndex():RDD[(T,Long)]

zipWithIndex操作将RDD中的元素和这个元素在RDD中的ID(索引号)组合成键/值对

zipWithUniqueId():RDD[(T,Long)]

zipWithUniqueId 操作将RDD中的元素和一个唯一ID组合成键/值对，该唯一ID生成算法如下:

• 每个分区中第一个元素的唯一ID值为: 该分区索引号
• 每个分区中第N个元素的唯一ID值为:(前一个元素的唯一ID值) + (该RDD总的分区数)

键值转换操作

partitionBy(partitioner: Partitioner):RDD[(K, V)]

partitionBy 操作根据partitioner函数生成新的ShuffleRDD

mapValues[U](f:(V) => U):RDD[(K, U)]

mapValues类似于map，只不过mapValues是针对[K,V]中的V值进行map操作

flatMapValues[U](f:(V) => TraverableOnce[U]):RDD[(K,U)]

flatMapValues 相对flatMap, flatMapValues是针对[K,V]中的V值进行flatMap操作

combineByKey[C](createCombiner:(V), mergeValue:(C, V), mergeCombiners)

combineByKey 操作用于将RDD[K,V]转换成RDD[K,C].这里V类型和C类型可以相同也可以不同

combineByKey 参数含义

• createCombiner: 组合器函数，用于将V类型转换成C类型，输入参数为RDD[K,V]中的V，输出为C
• mergeValue: 合并值函数，将一个C类型和一个V类型值合并成一个C类型，输入参数为(C,V),输出为C
• mergeCombiners: 合并组合器函数，用于将两个C类型值合并成一个C类型，输入参数为(C,C)，输出为C
• numPartitons: 结果RDD分区数，默认保持原有的分区数
• partitioner: 分区函数，默认为HashPartitioner
• mapSideCombine: 是否需要在Map端进行combine操作，类似于MapReduce中的combine,默认为true

foldByKey(zeroValue:V)(func:(V, V)=>V):RDD[(K,V)]

foldByKey 操作用于RDD[K, V]转换K将V做折叠、合并处理。其中参数zeroValue表示先根据映射函数将zeroValue应用于V
进行初始化V，再将映射函数应用于初始化后的V

reduceByKey(func:(V, V) => V):RDD[(K,V)]

reduceByKey操作用于将RDD[K,V]中每个K对应的V值根据映射函数来运算，其中参数numPartitions用于指定分区数，参数partitioner用于指定分区函数

reduceByLocally(func:(V, V) => V):Map[K,V]

reduceByLocally 和 reduceByKey 功能类似，不同的是，reduceByLocally运算结果映射到一个Map[K,V]中，而不是RDD[K,V]

groupByKey():RDD[(K, Iterable[V])]

groupByKey 操作用于将RDD[K,V]中每个K对应的V值合并到一个集合Iterable[V]中

cogroup[W](other:RDD[(K,V)]):RDD[(K,(Iterable[V], Iterable[W]))]

cogroup 相当于SQL中的全外关联，返回左右RDD中的记录，关联不上的为空

可传入的参数有1~3个RDD，参数numPartitons用于指定分区数，参数partitioner用于指定分区函数

join、fullOuterJoin、leftOuterJoin、rightOuterJoin

join、fullOuterJoin、leftOuterJoin、rightOuterJoin 都是针对RDD[K,V]中K值相等的连接操作

分别对应内连接、全连接、左连接和右连接。这些操作都调用cogroup进行实现，subtractByKey和基本操作subtract,只是subtractByKey针对的是键值操作

其中参数numPartions用于指定分区数，参数partitioner用于指定分区函数

控制操作

cache():RDD[T]

缓存

persist():RDD[]

persit(level:StorageLevel):RDD[T]

行动操作

集合标量行动操作

first(): T 表示返回RDD中的第一个元素，不排序

count(): Long 表示返回RDD中的元素个数

reduce(f:(T, T) => T):T

根据映射函数f，对RDD中的元素进行二元计算

collect(): Array[T]

表示将RDD转换成数组

take(num: Int): Array[T]

表示获取RDD中从0到num-1下标的元素，不排序

top(num: Int): Array[T]

表示从RDD中，按照默认(降序)或者指定的排序规则，返回前num个元素

takeOrdered(num: Int): Array[T]

和top类似，只不过以和top相反的顺序返回元素

aggregate[U](zeroValue: U)(seqOp: (U, T) => U, combOp: (U,U) => U)

用户聚合RDD中的元素，先使用seqOp将RDD中每个分区中的T类型聚合成U类型

再使用combOp将之前每个分区聚合后的U类型聚合成U类型，特别注意seqOp和combOp都会使用zeroValue的值，zeroValue的类型的为U

fold(zeroValue: T)(op: (T, T)=>T): Taggregate

fold是aggregate的简化，将aggregate中的seqOp和combOp使用同一个函数op

lookup(key: K): Seq[V]

lookup用于(K,V)类型的RDD，指定K值，返回RDD中该K对应的所有V值

countByKey(): Map[K, Long]

countByKey 统计RDD[K, V]中每个K的数量

foreach(f:(T) => Unit): Unit

foreach 遍历RDD，将函数f应用于每一个元素

要注意如果对RDD执行foreach，只会在Executor端有效，而并不是Driver端

foreachPartition(f:(Iterator[T]) => Unit): Unit

foreachPartition 和 foreach类似，只不过是对每一个分区使用f

sortBy[K](f:(T), ascending:Boolean, numPartitions):RDD[T]

sortBy根据给定的排序k函数将RDD中的元素进行排序

存储行动操作

saveAsTextFile(path: String): Util

saveAsTextFile 用于将RDD以文本文件的格式存储到文件系统中，codec参数可以指定压缩的类名

saveAsTextFile 用于将RDD以SequenceFile的文件格式保存到HDFS上

saveAsObjectFile(path: String): Util

saveAsObjectFile 用于将RDD中的元素序列化对象，存储到文件中，对于HDFS，默认采用SequenceFile保存

saveAsHadoop

saveAsHadoopFile 是将RDD存储在HDFS上的文件中

saveAsHadoopDataset(conf: JobConf): Unit

用于将RDD保存到除了HDFS的其他存储中，比如HBase

在JobConf中，通常需要关注或者设置5个参数: 文件的保存路径、key值的class类型、value值的class类型、RDD的输出格式以及压缩相关参数