spark.api.java

mapToPair

此函數(shù)會對一個RDD中的每個元素調(diào)用f函數(shù)，調(diào)用f函數(shù)后會進行一定的操作把每個元素都轉(zhuǎn)換成一個<K,V>類型的對象

• 源碼

  /**
   * Return a new RDD by applying a function to all elements of this RDD.
   */
  def mapToPair[K2, V2](f: PairFunction[T, K2, V2]): JavaPairRDD[K2, V2] = {
    def cm: ClassTag[(K2, V2)] = implicitly[ClassTag[(K2, V2)]]
    new JavaPairRDD(rdd.map[(K2, V2)](f)(cm))(fakeClassTag[K2], fakeClassTag[V2])
  }

• demo

val rdd: RDD[String] = sc.makeRDD(Seq("java", "scala", "rdd"))

    val javaPairRdd: JavaPairRDD[String, Int] = rdd.toJavaRDD().mapToPair(f => {
      (f, 1)
    })
    javaPairRdd.foreach(println)
    //(java,1)
    //(scala,1)
    //(rdd,1)

flatMapToPair

函數(shù)相對于mapToPair多了個flatMap的功能，它能夠?qū)⒃夭鸱趾笤谵D(zhuǎn)換成鍵值對

• 源碼

  /**
   *  Return a new RDD by first applying a function to all elements of this
   *  RDD, and then flattening the results.
   */
  def flatMapToPair[K2, V2](f: PairFlatMapFunction[T, K2, V2]): JavaPairRDD[K2, V2] = {
    def fn: (T) => Iterator[(K2, V2)] = (x: T) => f.call(x).asScala
    def cm: ClassTag[(K2, V2)] = implicitly[ClassTag[(K2, V2)]]
    JavaPairRDD.fromRDD(rdd.flatMap(fn)(cm))(fakeClassTag[K2], fakeClassTag[V2])
  }

• demo

 val flatMapPairRdd: JavaPairRDD[String, Int] = rdd.toJavaRDD().flatMapToPair(f => {
      val arrayList = new util.ArrayList[(String, Int)]()
      arrayList.add((f, 1))
      arrayList.iterator()
    })

    flatMapPairRdd.foreach(println)
    //(java,1)
    //(scala,1)
    //(rdd,1)

spark.api.scala

scala 無這兩個算子，以下是代替方案（麻煩），建議還是轉(zhuǎn)javaRdd吧
maptoPair = map
flatMapToPair = flatMap + map