mapValues与map算子、flatMapValues与flatMap算子的原理基本上相同。为什么没有把这两个算子放在transformation基本原理一中而单拎出来,是因为"transformation基本原理一"中的算子都属于RDD类,而mapValues、flatMapValues属于PairRDDFunctions类,作用在元素类型为(k,v)的RDD上,其返回RDD的元素类型也是(k,v)
看下mapValues和flatMapValues的源码
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
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组装进来