Spark-Core RDD转换算子-Value型

痛定思痛。 2023-06-22 10:25 21阅读 0赞

### 1、 map(func) ###

**作用**: 返回一个新的 RDD, 该 RDD 是由原 RDD 的每个元素经过函数转换后的值而组成. 就是对 RDD 中的数据做转换.

创建一个包含1-10的的 RDD，然后将每个元素\*2形成新的 RDD

scala > val rdd1 = sc.parallelize(1 to 10)
    // 得到一个新的 RDD, 但是这个 RDD 中的元素并不是立即计算出来的
    scala> val rdd2 = rdd1.map(_ * 2)

### 2、mapPartitions(func) ###

**作用**: 类似于map(func), 但是是**独立在每个分区**上运行.所以:`Iterator<T> => Iterator<U>`

**假设有N个元素，有M个分区，那么map的函数的将被调用N次,而mapPartitions被调用M次,一个函数一次处理所有分区。**

scala > val rdd1 = sc.parallelize(1 to 10)
    // 得到一个新的 RDD, 但是这个 RDD 中的元素并不是立即计算出来的
    scala> val rdd2 = rdd1.mapPartitions(_ * 2)
    scala> rdd2.collect
    res9: Array[Int] = Array(2, 4, 6, 8, 10, 12, 14, 16, 18, 20)

### 3、mapPartitionsWithIndex(func) ###

**作用**: 和mapPartitions(func)类似. 但是**会给func多提供一个Int值来表示分区的索引**. 所以func的类型是:

`(Int, Iterator<T>)=> Iterator<U>`

scala> val rdd1 = sc.parallelize(Array(10,20,30,40,50,60))
    scala> val rdd2 = rdd1.mapPartitionsWithIndex((index, items) => items.map((index, _)))
    scala> res2.collect
    res9: Array[(Int, Int)] = Array((0,10), (0,20), (0,30), (1,40), (1,50), (1,60))

##### （1）确定分区数 #####

override def defaultParallelism(): Int =
       scheduler.conf.getInt("spark.default.parallelism", totalCores)

##### （2）对元素分区 #####

// length: RDD 中数据的长度  numSlices: 分区数
    def positions(length: Long, numSlices: Int): Iterator[(Int, Int)] = {
     (0 until numSlices).iterator.map { i =>
       val start = ((i * length) / numSlices).toInt
       val end = (((i + 1) * length) / numSlices).toInt
       (start, end)
     }
    }
    seq match {
     case r: Range =>
       
     case nr: NumericRange[_] =>
       
     case _ =>
       val array = seq.toArray // To prevent O(n^2) operations for List etc
       positions(array.length, numSlices).map { case (start, end) =>
           array.slice(start, end).toSeq
       }.toSeq
    }

### 5、map和mapPartitions的区别 ###

（1） map()：**每次处理一条数据**。

（2） mapPartition()：**每次处理一个分区的数据**，这个分区的数据处理完后，**原 RDD 中该分区的数据才能释放，可能导致 OOM**。

### 6、flatMap(func) ###

**作用**: 类似于map，但是**每一个输入元素**可以被映射为 `0`或`多`个输出元素（所以func应该返回一个序列，而不是单一元素 `T => TraversableOnce[U]）`

创建一个元素为 1-5 的RDD，运用 flatMap创建一个新的 RDD，新的 RDD 为原 RDD 每个元素的 平方和三次方 来组成 1,1,4,8,9,27..

scala> val rdd1 = sc.parallelize(Array(1,2,3,4,5))
    scala> val rdd2 =rdd1.flatMap(x => Array(x * x, x * x * x))
    scala> rdd2.collect
    res14: Array[Int] = Array(1, 1, 4, 8, 9, 27, 16, 64, 25, 125)

### 7、glom() ###

**作用**: 将每一个分区的元素合并成一个数组，形成新的 RDD 类型是RDD\[Array\[T\]\]

scala> var rdd1 = sc.parallelize(Array(10,20,30,40,50,60), 4)
    scala> rdd1.glom.collect
    res2: Array[Array[Int]] = Array(Array(10), Array(20, 30), Array(40), Array(50, 60))

### 8、groupBy(func) ###

**作用**:按照func的返回值进行分组.

**func返回值作为 key, 对应的值放入一个迭代器中. 返回的 `RDD: RDD[(K,Iterable[T])`**

创建一个 RDD，按照元素的奇偶性进行分组

scala> val rdd1 = sc.makeRDD(Array(1, 3, 4, 20, 4, 5, 8))
    scala> val rdd2 = rdd1.groupBy(x => if(x % 2 == 1) "odd" else "even")
    scala> rdd2.collect
    res5: Array[(String, Iterable[Int])] = Array((even,CompactBuffer(4, 20, 4, 8)), (odd,CompactBuffer(1, 3, 5)))

### 9、filter（func） ###

**作用:** 过滤. 返回一个新的 RDD 是由func的`返回值为true`的那些元素组成

创建一个 RDD（由字符串组成），过滤出一个新 RDD（包含“xiao”子串）

scala> val rdd1 = sc.parallelize(Array("xiaoli", "laoli", "laowang", "xiaocang", "xiaojing", "xiaokong"))
    scala> val rdd2 = rdd1.filter(_.contains("xiao"))
    scala> rdd2.collect
    res4: Array[String] = Array(xiaoli, xiaocang, xiaojing, xiaokong)

### 10、sample（withReplacement，fraction，seed） ###

**作用**

（1）以指定的随机种子随机抽样出比例为`fraction`的数据，(抽取到的数量是: `size * fraction`). 需要注意的是得到的结果并不能保证准确的比例

（2）withReplacement表示是**抽出的数据是否放回**

true为有放回的抽样

false为无放回的抽样

**true放回表示数据有可能会被重复抽取到，是 true， 则fraction大于等于0就可以了**

**false 则不可能重复抽取到. 如果是false, 则fraction必须是:\[0,1\]**

（3）seed用于指定随机数生成器种子。 一般用默认的, 或者传入当前的时间戳

**（4）不放回抽样**

scala> val rdd1 = sc.parallelize(1 to 10)
    
    scala> rdd1.sample(false, 0.5).collect
    res15: Array[Int] = Array(1, 3, 4, 7)

**（5）放回抽样**

scala> val rdd1 = sc.parallelize(1 to 10)
    scala> rdd1.sample(true, 2).collect
    res25: Array[Int] = Array(1, 1, 2, 3, 3, 4, 4, 5, 5, 5, 5, 5, 6, 6, 7, 7, 8, 8, 9)

### 11、distinct（\[numTasks\]） ###

**作用**:对 RDD 中元素执行去重操作. 参数表示任务的数量.**默认值和分区数保持一致**.

scala> val rdd1 = sc.parallelize(Array(10,10,2,5,3,5,3,6,9,1))
    
    scala> rdd1.distinct().collect
    res29: Array[Int] = Array(6, 10, 2, 1, 3, 9, 5)

**本质上是reduceByKey**

/**
       * Return a new RDD containing the distinct elements in this RDD.
       */
      def distinct(numPartitions: Int)(implicit ord: Ordering[T] = null): RDD[T] = withScope {
        map(x => (x, null)).reduceByKey((x, y) => x, numPartitions).map(_._1)
      }

### 12 、coalesce（numPartitions） ###

**作用**: `缩减分区数`到指定的数量，用于大数据集过滤后，提高小数据集的执行效率。

scala> val rdd1 = sc.parallelize(0 to 100, 5)
    
    scala> rdd1.partitions.length
    res3: Int = 5
    
    // 减少分区的数量至 2 
    scala>val rdd2= rdd1.coalesce(2)
    
    scala> rdd2.partitions.length
    res4: Int = 2

##### 注意: #####

`第二个参数表示是否shuffle,` **如果不传或者传入的为false`,** 则表示不进行shuffer`,** 此时分区数减少有效, 增加分区数无效.\*\*

### 13、repartition（numPartitions） ###

**作用:** 根据新的分区数, 重新 shuffle 所有的数据, 这个操作总会通过网络（跨分区操作）.

**新的分区数相比以前可以多, 也可以少**，一般用于增加分区，语义清晰

scala> val rdd1 = sc.parallelize(0 to 100, 5)
    
    scala> val rdd2 = rdd1.repartition(3)
    
    scala> res2.partitions.length
    res4: Int = 3
    
    scala> val rdd3 = rdd1.repartition(10)
    
    scala> rdd3.partitions.length
    res5: Int = 10

##### `coalasce和repartition的区别` #####

（1） coalesce重新分区，可以选择是否进行shuffle过程。由参数shuffle: Boolean = false/true决定。

（2）`repartition实际上是调用的的coalesce`，进行shuffle

def repartition(numPartitions: Int)(implicit ord: Ordering[T] = null): RDD[T] = withScope {
    coalesce(numPartitions, shuffle = true)
    }

（3）如果是减少分区, 尽量避免shuffle

### 14、sortBy（func，\[ascending\],\[numTasks\]） ###

**作用:** 使用func先对数据进行处理，按照处理后结果排序，**默认为正序**。

scala> val rdd1 = sc.parallelize(Array(1,3,4,10,4,6,9,20,30,16))
    
    scala> rdd1.sortBy(x => x).collect
    res17: Array[Int] = Array(1, 3, 4, 4, 6, 9, 10, 16, 20, 30)
    
    scala> rdd1.sortBy(x => x, true).collect
    res18: Array[Int] = Array(1, 3, 4, 4, 6, 9, 10, 16, 20, 30)
    
    // 不用正序
    scala> rdd1.sortBy(x => x, false).collect
    res19: Array[Int] = Array(30, 20, 16, 10, 9, 6, 4, 4, 3, 1)

### 15、pipe（command，\[envVars\]） ###

**作用**: 管道，`针对每个分区`，把 RDD 中的每个数据通过管道传递给shell命令或脚本，返回输出的RDD。

`一个分区执行一次这个命令`. `如果只有一个分区, 则执行一次命令`.

**注意:**

脚本要放在 worker 节点可以访问到的位置

（1）创建一个脚本文件pipe.sh

echo "hello"
    while read line;do
        echo ">>>"$line
    done

（2）创建只有 1 个分区的RDD

scala> val rdd1 = sc.parallelize(Array(10,20,30,40), 1)
    
    scala> rdd1.pipe("./pipe.sh").collect
    res1: Array[String] = Array(hello, >>>10, >>>20, >>>30, >>>40)

（3）创建有 2 个分区的 RDD

scala> val rdd1 = sc.parallelize(Array(10,20,30,40), 2)
    
    scala> rdd1.pipe("./pipe.sh").collect
    res2: Array[String] = Array(hello, >>>10, >>>20, hello, >>>30, >>>40)

#### 每个分区执行一次脚本, 但是每个元素算是标准输入中的一行 ####