Spark操作——转换操作(二)

ゝ一世哀愁。 2023-03-12 07:11 18阅读 0赞

*  **基础转换操作**
 *  **键值转换操作**

### 基础转换操作 ###

*  mapPartitions\[U\](f:(Iterator\[T\]) => Iterator\[U\], preservesPartitioning: Boolean=false): RDD\[U\]

mapPartitions操作和map操作类似，区别在于映射函数的参数由RDD中的每一个元素变成了RDD中的每一个分区的迭代器，可以理解为map操作中对RDD中的所有元素应用映射函数，mapPartitions操作是对RDD中每个分区的元素应用映射函数，相当于按分区进行分组。参数preservesPartitioning标识是否保留父RDD的分区信息。

该操作的应用场景适合那些在映射过程中需要频繁创建额外对象，效率比map高。举例，将RDD中的元素通过JDBC连接数据库并写入，map操作就需要为每个元素建立一个连接，开销很大，而mapPartitions只需要对每个分区建立一个连接。

scala> var rdd = sc.parallelize(1 to 10, 4)
    rdd: org.apache.spark.rdd.RDD[Int] = ParallelCollectionRDD[52] at parallelize at <console>:24
    
    # 将rdd中的每个分区元素累加
    scala> var rdd2 = rdd.mapPartitions(x => {var result = List[Int](); var i=0; while(x.hasNext){i+=x.next;}; result.::(i).iterator})
    rdd2: org.apache.spark.rdd.RDD[Int] = MapPartitionsRDD[68] at mapPartitions at <console>:25
    
    scala> rdd2.collect
    res44: Array[Int] = Array(3, 12, 13, 27)

*  mapPartitionsWithIndex\[U\](f:(Int, Iterator\[T\]) => Iterator\[U\], preservesPartitioning: Boolean=false):RDD\[U\]

该操作与mapPartitions类似，区别在于输入参数多了一个分区索引

scala> var rdd = sc.parallelize(1 to 10, 4)
    rdd: org.apache.spark.rdd.RDD[Int] = ParallelCollectionRDD[52] at parallelize at <console>:24
    
    # index为分区索引，将rdd中的每个分区元素累加，并在结果前面天啊及分区索引
    scala> var rdd2 = rdd.mapPartitionsWithIndex((index,x) => {var result = List[String](); var i=0; while(x.hasNext){i+=x.next;}; result.::(index + "|" + i).iterator})
    rdd2: org.apache.spark.rdd.RDD[String] = MapPartitionsRDD[69] at mapPartitionsWithIndex at <console>:25
    
    scala> rdd2.collect
    res48: Array[String] = Array(0|3, 1|12, 2|13, 3|27)

*  zip\[U\](other:RDD\[U\]):RDD\[T,U\]

将两个RDD组合成Key/Value形势的RDD，默认两个RDD具有相同的分区数和元素个数，否则会报错。

scala> var rdd1 = sc.makeRDD(1 to 3, 2)
    rdd1: org.apache.spark.rdd.RDD[Int] = ParallelCollectionRDD[70] at makeRDD at <console>:24
    
    scala> var rdd2 = sc.makeRDD(4 to 6, 2)
    rdd2: org.apache.spark.rdd.RDD[Int] = ParallelCollectionRDD[71] at makeRDD at <console>:24
    
    scala> var rdd3 = sc.makeRDD(7 to 9, 1)
    rdd3: org.apache.spark.rdd.RDD[Int] = ParallelCollectionRDD[72] at makeRDD at <console>:24
    
    scala> rdd1.zip(rdd2).collect
    res49: Array[(Int, Int)] = Array((1,4), (2,5), (3,6))
    
    scala> rdd1.zip(rdd3).collect
    java.lang.IllegalArgumentException: Can't zip RDDs with unequal numbers of partitions: List(2, 1)
      at org.apache.spark.rdd.ZippedPartitionsBaseRDD.getPartitions(ZippedPartitionsRDD.scala:57)
      at org.apache.spark.rdd.RDD$$anonfun$partitions$2.apply(RDD.scala:253)
      at org.apache.spark.rdd.RDD$$anonfun$partitions$2.apply(RDD.scala:251)
      at scala.Option.getOrElse(Option.scala:121)
      at org.apache.spark.rdd.RDD.partitions(RDD.scala:251)
      at org.apache.spark.SparkContext.runJob(SparkContext.scala:2099)
      at org.apache.spark.rdd.RDD$$anonfun$collect$1.apply(RDD.scala:945)
      at org.apache.spark.rdd.RDDOperationScope$.withScope(RDDOperationScope.scala:151)
      at org.apache.spark.rdd.RDDOperationScope$.withScope(RDDOperationScope.scala:112)
      at org.apache.spark.rdd.RDD.withScope(RDD.scala:363)
      at org.apache.spark.rdd.RDD.collect(RDD.scala:944)
      ... 49 elided

*  zipPartitions\[B,V\](rdd2:RDD\[B\])(f:(Iterator\[T\], Iterator\[B\]) => Iterator\[V\]):RDD\[V\]
 *  zipPartitions\[B,V\](rdd2:RDD\[B\], preservesPartitioning:Boolean)(f:(Iterator\[T\], Iterator\[B\]) => Iterator\[V\]):RDD\[V\]
 *  zipPartitions\[B,C,V\](rdd2:RDD\[B\], rdd3:RDD\[C\])(f:(Iterator\[T\], Iterator\[B\], Iterator\[C\]) => Iterator\[V\]):RDD\[V\]
 *  zipPartitions\[B,C,V\](rdd2:RDD\[B\], rdd3:RDD\[C\], preservesPartitioning:Boolean)(f:(Iterator\[T\], Iterator\[B\], Iterator\[C\]) => Iterator\[V\]):RDD\[V\]
 *  zipPartitions\[B,C,D,V\](rdd2:RDD\[B\], rdd3:RDD\[C\], rdd4:RDD\[D\])(f:(Iterator\[T\], Iterator\[B\], Iterator\[C\], Iterator\[D\]) => Iterator\[V\]):RDD\[V\]
 *  zipPartitions\[B,C,D,V\](rdd2:RDD\[B\], rdd3:RDD\[C\], rdd4:RDD\[D\], preservesPartitioning:Boolean)(f:(Iterator\[T\], Iterator\[B\], Iterator\[C\], Iterator\[D\]) => Iterator\[V\]):RDD\[V\]

zipPartitions操作将多个RDD按照分区组合为新的RDD，默认需要组合的RDD具有相同的分区，但对于每个分区内的元素个数并不要求一致。

*  zipWithIndex(): RDD\[(T, Long)\]

zipWithIndex操作将RDD中的元素和该元素在RDD中的索引号组合成键值对。

scala> rdd1.zipWithIndex().collect
    res51: Array[(Int, Long)] = Array((1,0), (2,1), (3,2))

*  zipWithUniqueId(): RDD\[(T, Long)\]

zipWithUniqueId操作将RDD中的元素与一个唯一的ID组合成键值对，该唯一ID的算法如下：

1.  每个分区第一个元素的ID为：该分区索引号
2.  每个分区第N个元素的ID为：前一个元素的唯一ID值 + 该RDD的分区数

scala> rdd1.zipWithUniqueId().collect
    res52: Array[(Int, Long)] = Array((1,0), (2,1), (3,3))

参考:

\[1\] 郭景瞻. 图解Spark:核心技术与案例实战\[M\]. 北京:电子工业出版社, 2017.