spark转换算子-flatMap（）扁平化

1）函数签名：

def flatMap[U:ClassTag](f:T=>TraversableOnce[U]):RDD[U]

2）功能说明

与map操作类似，将RDD中的每一个元素通过应用f函数依次转换为新的元素，并封装到RDD中。

区别：在flatMap操作中，f函数的返回值是一个集合，并且会将每一个该集合中的元素拆分出来放到新的RDD中。

3）需求操作：创建一个集合，集合里面存储的还是子集合，把所有子集合中数据取出放入到一个大的集合中。

4）具体实现：

package com.huc.Spark1.value

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

object Test04_FlatMap {
  def main(args: Array[String]): Unit = {
    //1.创建SparkConf并设置App名称
    val conf: SparkConf = new SparkConf().setAppName("SparkCore").setMaster("local[*]")

    //2.创建SparkContext，该对象是提交Spark App的入口
    val sc: SparkContext = new SparkContext(conf)

    //3.使用Scala进行spark编程
    val rdd: RDD[List[Int]] = sc.makeRDD(List(List(1, 2, 3, 4), List(3, 4, 5)), 2)
    val value: RDD[Int] = rdd.flatMap(list => list)
    println(value.collect().mkString(","))
    value.collect().foreach(println)
    // flatMap 不会改变原有分区
    println(rdd.mapPartitionsWithIndex((index, datas) => datas.map(data => (index, data))).collect().mkString(","))
    println(value.mapPartitionsWithIndex((index, datas) => datas.map(data => (index, data))).collect().mkString(","))

    //4.关闭连接
    sc.stop()

  }
}

package com.huc.Spark.value

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

object Test04_flatMap {
  def main(args: Array[String]): Unit = {

    //1.创建SparkConf并设置App名称
    val conf: SparkConf = new SparkConf().setAppName("SparkCore").setMaster("local[*]")

    //2.创建SparkContext，该对象是提交Spark App的入口
    val sc: SparkContext = new SparkContext(conf)

    //3.使用Scala进行spark编程
    // 创建一个RDD
    val list1: RDD[List[Int]] = sc.makeRDD(List(List(1, 2), List(3, 4), List(5, 6), List(7)), 2)

    // 把所有子集合中的数据取出来放到一个大的集合中
    val list2: RDD[Int] = list1.flatMap((list: List[Int]) => list)
    list2.collect().foreach(println)
    println(list2.collect().toList)

    //4.关闭连接
    sc.stop()

  }
}