前言
mapValues与map算子、flatMapValues与flatMap算子的原理基本上相同。为什么没有把这两个算子放在transformation基本原理一中而单拎出来,是因为"transformation基本原理一"中的算子都属于RDD类,而mapValues、flatMapValues属于PairRDDFunctions类,作用在元素类型为(k,v)的RDD上,其返回RDD的元素类型也是(k,v)
看下mapValues和flatMapValues的源码
源码
mapValues
def mapValues[U](f: V => U): RDD[(K, U)] = self.withScope {
// 对入参f进行检查,如是否可序列化等
val cleanF = self.context.clean(f)
// 创建MapPartitionsRDD对象
new MapPartitionsRDD[(K, U), (K, V)](self,
(context, pid, iter) => iter.map { case (k, v) => (k, cleanF(v)) },
preservesPartitioning = true)
}
mapValues算子是一个transformation算子,会生成一个MapPartitionsRDD对象, MapPartitionsRDD原理请参考MapPartitionsRDD基本原理。重点关注构建
MapPartitionsRDD对象时传入的第二个参数:
(context, pid, iter) => iter.map { case (k, v) => (k, cleanF(v)) }
先看下入参:context和pid暂时不用关注,iter表示父RDD分区迭代器。
方法的实现:调用map方法,注意这个map不是spark算子,而是scala迭代器的map方法。
再看map方法的入参:(k, v) => (k, cleanF(v))。因为mapValues是作用在(k,v)类型的RDD上,所以入参为(k,v)。出参也是(k,v)。入参的v的类型是V,出参v的类型是U
flatMapValues
def flatMapValues[U](f: V => TraversableOnce[U]): RDD[(K, U)] = self.withScope {
val cleanF = self.context.clean(f)
new MapPartitionsRDD[(K, U), (K, V)](self,
(context, pid, iter) => iter.flatMap { case (k, v) =>
cleanF(v).map(x => (k, x))
},
preservesPartitioning = true)
}
基本原理同mapValues算子一样,不做过多赘述。
解释下(k, v) => cleanF(v).map(x => (k, x)):
入参(k,v),即父RDD中元素的类型
cleanF(v).map(x => (k, x)):先对v做一个转换,生成一个迭代器,再调用迭代器的map方法,把k组装进来