RDD转换算子：【mapValues、mapPartitions】

---

1、mapValues算子

功能

针对二元组KV类型的RDD，对RDD中每个元素的Value进行map处理，结果放入一个新的RDD中

语法

def mapValues(self: RDD[Tuple[K,V]], f: (V) -> U) -> RDD[Tuple[K,U]]

举例

如何构造sc对象并创建RDD 参考文章：
【Spark中创建RDD的两种方式】Spark中如何获取sc对象、以及创建RDD的两种方式

rdd_kv = sc.parallelize([('laoda',11),('laoer',22),('laosan',33),('laosi',44)], numSlices=2)
# 拿到年龄，并加一
rsRdd = rdd_kv.mapValues(lambda age: age + 1)
rsRdd.foreach(lambda x:print(x))

2、mapPartitions算子

功能

对RDD每个分区的数据进行操作，将每个分区的数据进行map转换，将转换的结果放入新的RDD中

与 map 的区别：每个分区会调用一次（一个分区一个分区的数据进行处理，而map是一条一条数据进行处理），将这个分区的数据放入内存，性能比map更好，优化型算子，注意更容易出现内存溢出

语法

def mapPartitions(self: RDD[T], f: Iterable[T] -> Iterable[U] ) -> RDD[U]

• 优点：性能快、节省外部连接资源
• 缺点：如果单个分区的数据量较大，容易出现内存溢出
• 场景：
  • 数据量不是特别大，需要提高性能【将整个分区的数据放入内存】
  • 需要构建外部资源时【基于每个分区构建一份资源】

举例

# 使用mapPartitions：对每个分区进行处理
def map_partition(part):rs = [i * 2 for i in part]return rs# 每个分区会调用一次：将这个分区的数据放入内存，性能比map更好，优化型算子，注意更容易出现内存溢出
map_part_rdd = input_rdd.mapPartitions(lambda part: map_partition(part))