标签：map 基本原理 mapValues RDD 算子 cleanF transformation MapPartitionsRDD

前言

        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组装进来

标签：map,基本原理,mapValues,RDD,算子,cleanF,transformation,MapPartitionsRDD
来源： https://blog.csdn.net/huyang0101/article/details/122128670

本站声明： 1. iCode9 技术分享网（下文简称本站）提供的所有内容，仅供技术学习、探讨和分享；
2. 关于本站的所有留言、评论、转载及引用，纯属内容发起人的个人观点，与本站观点和立场无关；
3. 关于本站的所有言论和文字，纯属内容发起人的个人观点，与本站观点和立场无关；
4. 本站文章均是网友提供，不完全保证技术分享内容的完整性、准确性、时效性、风险性和版权归属；如您发现该文章侵犯了您的权益，可联系我们第一时间进行删除；
5. 本站为非盈利性的个人网站，所有内容不会用来进行牟利，也不会利用任何形式的广告来间接获益，纯粹是为了广大技术爱好者提供技术内容和技术思想的分享性交流网站。