- Value 类型
- map
- val mapRDD: RDD[Int] = rdd.map(_*2)
- 对传入的数据，一个一个的进行转换，再返回给结果集
- mapPartitions
- val mpRDD: RDD[Int] = rdd.mapPartitions(iter => { iter.map(_ * 2) })
- 一个分区一个分区的数据进行转换，传入迭代器，返回迭代器，再返回给结果集。迭代器数据量大小不做控制
- map 和 mapPartitions 的区别：
数据处理角度：
Map 算子是分区内一个数据一个数据的执行，类似于串行操作。而 mapPartitions 算子是以分区为单位进行批处理操作
功能的角度：
Map 算子主要目的将数据源中的数据进行转换和改变。但是不会减少或增多数据。
MapPartitions 算子需要传递一个迭代器，返回一个迭代器，没有要求的元素的个数保持不变，
所以可以增加或减少数据
性能的角度：
Map 算子因为类似于串行操作，所以性能比较低，而是 mapPartitions 算子类似于批处理，所以性能较高
但是 mapPartitions 算子会长时间占用内存，那么这样会导致内存可能不够用，出现内存溢出的错误。所以在内存有限的情况下，不推荐使用。使用 map 操作 - mapPartitionsWithIndex
- mapPartitionsWithIndex在mapPartitions基础上加上了分区index
- val mpiRDD = rdd.mapPartitionsWithIndex((index,iter) => { iter.map(num => { (index,num) } ）} )
- flatMap
val flatRDD:RDD[String] = rdd.flatMap( s => { s.split(” “) })
Hello
Scala
Hello
Spark
如果使用rdd.map( s => { s.split(” “) })，会发现打印的结果是
[Ljava.lang.String;@f1a45f8
[Ljava.lang.String;@5edf2821
所以切割等扁平映射操作，选用flatMap- val flatRDD:RDD[String] = rdd.flatMap( s => { s.split(” “) })
- 扁平映射
- glom
- 将同一个分区的数据直接转换为相同类型的内存数组进行处理
- List[Int] => Array[Int]
- groupBy
- val groupRDD:RDD[(Int,Iterable[Int])] = rdd.groupBy(num % 2)
- 会输出一个RDD[(K, Iterable[T])]
(0,CompactBuffer(2, 4))
(1,CompactBuffer(1, 3))
- filter
- 将数据根据指定的规则进行筛选过滤，符合规则的数据保留，不符合规则的数据丢弃。
- val dataRDD1 = dataRDD.filter(_%2 == 0)
- sample
- rdd.sample(true,0.5,101)
// sample算子需要传递三个参数
// 1. 第一个参数表示，抽取数据后是否将数据放回 true（放回），false (丢弃)
// 2. 第二个参数表示
// 如果抽取不放回的场合：数据源中每条数据被抽取的概率，基准值的概念 // 如果抽取放回的场合：表示数据源中的每条数据被抽取的可能次数 // 3. 第三个参数表示，抽取数据时随机算法的种子 // 如果不传递第三个参数，那么使用的是当前系统时间
- rdd.sample(true,0.5,101)
- distinct
- 将数据集中重复的数据去重
- val dataRDD1 = dataRDD.distinct()
- coalesce
- 缩减分区
- val newRDD = rdd.coalesce(2, true) 默认为false,不进行shuffle操作，将分区数量改小
- repartition
- repartition 操作内部其实执行的是 coalesce 操作，参数 shuffle 的默认值为 true
- sortBy
- 该操作用于排序数据。在排序之前，可以将数据通过 f 函数进行处理，之后按照 f 函数处理的结果进行排序，默认为升序排列。排序后新产生的 RDD 的分区数与原 RDD 的分区数一致。中间存在 shuffle 的过程
- val dataRDD1 = dataRDD.sortBy(num=>num, false, 4) 默认升序排序，修改分区数量为4，发生shuffle操作
- map
- 双value类型
- intersection
- def intersection(other: RDD[T]): RDD[T]
- val dataRDD = dataRDD1.intersection(dataRDD2)
- 求交集
- union
- def union(other: RDD[T]): RDD[T]
- val dataRDD = dataRDD1.union(dataRDD2)
- 求并集
- subtract
- def subtract(other: RDD[T]): RDD[T]
- val dataRDD = dataRDD1.subtract(dataRDD2)
- 求差集
- zip
- def zip[U: ClassTag](other: RDD[U]): RDD[(T, U)]
- val dataRDD = dataRDD1.zip(dataRDD2)
- 将两个 RDD 中的元素，以键值对的形式进行合并。其中，键值对中的 Key 为第 1 个 RDD
- 中的元素，Value 为第 2 个 RDD 中的相同位置的元素
- Key-Value类型
- partitionBy
- def partitionBy(partitioner: Partitioner): RDD[(K, V)]
- 将数据按照指定 Partitioner 重新进行分区。Spark 默认的分区器是 HashPartitioner
- val rdd2: RDD[(Int, String)] = rdd.partitionBy(new HashPartitioner(2))
- reduceByKey
- def reduceByKey(func: (V, V) => V): RDD[(K, V)]
- def reduceByKey(func: (V, V) => V, numPartitions: Int): RDD[(K, V)]
- 可以将数据按照相同的 Key 对 Value 进行聚合
- val dataRDD2 = dataRDD1.reduceByKey(_+_)
- groupByKey
- def groupByKey(): RDD[(K, Iterable[V])]
- def groupByKey(numPartitions: Int): RDD[(K, Iterable[V])]
- def groupByKey(partitioner: Partitioner): RDD[(K, Iterable[V])]
- 将数据源的数据根据 key 对 value 进行分组
- val dataRDD2 = dataRDD1.groupByKey()
- val dataRDD3 = dataRDD1.groupByKey(2)
- val dataRDD4 = dataRDD1.groupByKey(new HashPartitioner(2))
- reduceByKey 和 groupByKey 的区别：
- 从 shuffle 的角度：reduceByKey 和 groupByKey 都存在 shuffle 的操作，但是 reduceByKey
- 可以在 shuffle 前对分区内相同 key 的数据进行预聚合（combine）功能，这样会减少落盘的
- 数据量，而 groupByKey 只是进行分组，不存在数据量减少的问题，reduceByKey 性能比较
- 高。
- 从功能的角度：reduceByKey 其实包含分组和聚合的功能。GroupByKey 只能分组，不能聚
- 合，所以在分组聚合的场合下，推荐使用 reduceByKey，如果仅仅是分组而不需要聚合。那
- 么还是只能使用 groupByKey
- aggregateByKey
- def aggregateByKey[U: ClassTag](zeroValue: U)(seqOp: (U, V) => U,combOp: (U, U) => U): RDD[(K, U)]
- 将数据根据不同的规则进行分区内计算和分区间计算
- dataRDD1.aggregateByKey(0)(_+_,_+_)
// 1. 第一个参数列表中的参数表示初始值
// 2. 第二个参数列表中含有两个参数
// 2.1 第一个参数表示分区内的计算规则
// 2.2 第二个参数表示分区间的计算规则
- foldByKey
- def foldByKey(zeroValue: V)(func: (V, V) => V): RDD[(K, V)]
- 当分区内计算规则和分区间计算规则相同时，aggregateByKey 就可以简化为 foldByKey
- combineByKey
- def combineByKey[C](
- createCombiner: V => C,
- mergeValue: (C, V) => C,
- mergeCombiners: (C, C) => C): RDD[(K, C)]
- 最通用的对 key-value 型 rdd 进行聚集操作的聚集函数（aggregation function）。类似于
- aggregate()，combineByKey()允许用户返回值的类型与输入不一致
- val combineRdd: RDD[(String, (Int, Int))] = input.combineByKey(
- (_, 1),
- (acc: (Int, Int), v) => (acc._1 + v, acc._2 + 1),
- (acc1: (Int, Int), acc2: (Int, Int)) => (acc1._1 + acc2._1, acc1._2 + acc2._2))
- reduceByKey、foldByKey、aggregateByKey、combineByKey 的区别：
reduceByKey: 相同 key 的第一个数据不进行任何计算，分区内和分区间计算规则相同
FoldByKey: 相同 key 的第一个数据和初始值进行分区内计算，分区内和分区间计算规则相同
AggregateByKey：相同 key 的第一个数据和初始值进行分区内计算，分区内和分区间计算规则可以不相同
CombineByKey:当计算时，发现数据结构不满足要求时，可以让第一个数据转换结构。分区内和分区间计算规则不相同 - sortByKey
- def sortByKey(ascending: Boolean = true, numPartitions: Int = self.partitions.length): RDD[(K, V)]
- val sortRDD1: RDD[(String, Int)] = dataRDD1.sortByKey(true)
- val sortRDD1: RDD[(String, Int)] = dataRDD1.sortByKey(false)
- join
- def join[W](other: RDD[(K, W)]): RDD[(K, (V, W))]
- 在类型为(K,V)和(K,W)的 RDD 上调用，返回一个相同 key 对应的所有元素连接在一起的(K,(V,W))的 RDD
- rdd.join(rdd1).collect().foreach(println)
- leftOuterJoin
- def leftOuterJoin[W](other: RDD[(K, W)]): RDD[(K, (V, Option[W]))]
- 类似于 SQL 语句的左外连接
- val rdd: RDD[(String, (Int, Option[Int]))] = dataRDD1.leftOuterJoin(dataRDD2)
- rightOuterJoin
- 类似于 SQL 语句的右外连接
- cogroup
- 在类型为(K,V)和(K,W)的 RDD 上调用，返回一个(K,(Iterable<V>,Iterable<W>))类型的 RDD
- val value: RDD[(String, (Iterable[Int], Iterable[Int]))] = dataRDD1.cogroup(dataRDD2)
- (a,(CompactBuffer(1),CompactBuffer(4)))
- (b,(CompactBuffer(2),CompactBuffer(5)))
- (c,(CompactBuffer(3),CompactBuffer(6, 7)))
- partitionBy
作者 admin
张宴银，大数据开发工程师