Spark中filter、map、flatMap、union、groupByKey、reduceByKey等共享变量示例方法记录

以你之姓@ 2021-09-28 15:30 503阅读 0赞

Spark提供了两种创建RDD的方式：读取外部数据集，以及在驱动器程序中对一个集合进行并行化。

在驱动器程序中对一个集合进行并行化的方式有两种：parallelize()和makeRDD()。

创建并行集合的一个重要参数，是slices的数目（例子中是numMappers），它指定了将数据集切分为几份

一般来说，Spark会尝试根据集群的状况，来自动设定slices的数目。当让，也可以手动的设置它，通过parallelize方法的第二个参数。

*  **map()、filter()、join() 等算子称为 Transformation，它们输入一个或多个 RDD，输出一个 RDD。**
 *  **collect()、count()、save() 等算子称为 Action，它们通常是将数据收集起来返回**

**map(func)** :返回一个新的分布式数据集，由每个原元素经过func函数转换后组成   
spark shell本地测试：

val numbers = Array(1, 2, 3, 4, 5)
        val numberRDD = sc.parallelize(numbers, 1)  
        val multipleNumberRDD = numberRDD.map ( num => num * 2 )
        multipleNumberRDD.foreach ( num => println(num) )

![image_1arhvvo24sjq5921hk81ile1ob19.png][]

**filter(func)** : 返回一个新的数据集，由经过func函数后返回值为true的原元素组成

val numbers = Array(1, 2, 3, 4, 5, 6, 7, 8, 9, 10)
        val numberRDD = sc.parallelize(numbers, 1)
        val evenNumberRDD = numberRDD.filter { num => num % 2 == 0 }

![image_1ari04fp91q8sdfb1b8r66tj4mm.png][]

**flatMap(func)** : 类似于map，但是每一个输入元素，会被映射为0到多个输出元素（因此，func函数的返回值是一个Seq，而不是单一元素）

val lineArray = Array("hello you", "hello me", "hello world")  
        val lines = sc.parallelize(lineArray, 1)
        val words = lines.flatMap { line => line.split(" ") }   
        words.foreach { word => println(word) }

![image_1ari0mqic1qmq1qvl1q2f1sq91dl13.png][]

**union(otherDataset)** : 返回一个新的数据集，由原数据集和参数联合而成   
**groupByKey(\[numTasks\])** :在一个由（K,V）对组成的数据集上调用，返回一个（K，Seq\[V\])对的数据集。注意：默认情况下，使用8个并行任务进行分组，你可以传入numTask可选参数，根据数据量设置不同数目的Task

val scoreList = Array(Tuple2("class1", 80), Tuple2("class2", 75),Tuple2("class1", 90), Tuple2("class2", 60))
        val scores = sc.parallelize(scoreList, 1)  
        val groupedScores = scores.groupByKey() 
         groupedScores.foreach(score => { 
          println(score._1); 
          score._2.foreach { singleScore => println(singleScore) };
          println("=============================")  })

![image_1ari0pnrr31b8pkgn141f1p661g.png][]

**reduceByKey(func, \[numTasks\])** : 在一个（K，V)对的数据集上使用，返回一个（K，V）对的数据集，key相同的值，都被使用指定的reduce函数聚合到一起。和groupbykey类似，任务的个数是可以通过第二个可选参数来配置的。在实际开发中，能使reduceByKey实现的就不用groupByKey

val scoreList = Array(Tuple2("class1", 80), Tuple2("class2", 75),Tuple2("class1", 90), Tuple2("class2", 60))
        val scores = sc.parallelize(scoreList, 1)  
        val totalScores = scores.reduceByKey(_ + _)  
        totalScores.foreach(classScore => println(classScore._1 + ": " + classScore._2))

![image_1ari0vb9ie2f1sb7ehc13tvl411t.png][]

**join(otherDataset, \[numTasks\])** :在类型为（K,V)和（K,W)类型的数据集上调用，返回一个（K,(V,W))对，每个key中的所有元素都在一起的数据集

val studentList = Array(
            Tuple2(1, "leo"),
            Tuple2(2, "jack"),
            Tuple2(3, "tom"));
       val scoreList = Array(
            Tuple2(1, 100),
            Tuple2(2, 90),
            Tuple2(3, 60));
        val students = sc.parallelize(studentList);
        val scores = sc.parallelize(scoreList);
        val studentScores = students.join(scores)  
        studentScores.foreach(studentScore => { 
          println("student id: " + studentScore._1);
          println("student name: " + studentScore._2._1)
          println("student socre: " + studentScore._2._2)  
          println("=======================================")  
        })

![image_1ari12cb8122f11tv1ufv14ao1i4n2a.png][]

**groupWith(otherDataset, \[numTasks\])** : 在类型为（K,V)和(K,W)类型的数据集上调用，返回一个数据集，组成元素为（K, **Seq\[V\], Seq\[W\]) Tuples**。这个操作在其它框架，称为CoGroup   
**cartesian(otherDataset)** : 笛卡尔积。但在数据集T和U上调用时，返回一个(T，U）对的数据集，所有元素交互进行笛卡尔积。   
**repartition（）**：重新分区，当数据处理到最后剩下很少的数据集时，可以使用repartition（）进行重新分区

**reduce(func)** : 通过函数func聚集数据集中的所有元素。Func函数接受2个参数，返回一个值。这个函数必须是关联性的，确保可以被正确的并发执行

val numberArray = Array(1, 2, 3, 4, 5, 6, 7, 8, 9, 10)
        val numbers = sc.parallelize(numberArray, 1)  
        val sum = numbers.reduce(_ + _)  
        println(sum)
    //结果55

**collect()** : 在Driver的程序中，以数组的形式，返回数据集的所有元素。这通常会在使用filter或者其它操作后，返回一个足够小的数据子集再使用

val numberArray = Array(1, 2, 3, 4, 5, 6, 7, 8, 9, 10)
        val numbers = sc.parallelize(numberArray, 1)  
        val doubleNumbers = numbers.map { num => num * 2 }  
        val doubleNumberArray = doubleNumbers.collect()
        for(num <- doubleNumberArray) {
          println(num)  
        }
    //结果
    2
    4
    6
    8
    10
    12
    14
    16
    18
    20

**count()** : 返回数据集的元素个数

val numberArray = Array(1, 2, 3, 4, 5, 6, 7, 8, 9, 10)
        val numbers = sc.parallelize(numberArray, 1)  
        val count = numbers.count()
        println(count) 
    //结果 10

**take(n)** : 返回一个数组，由数据集的前n个元素组成。注意，这个操作目前并非在多个节点上，并行执行，而是Driver程序所在机器，单机计算所有的元素(Gateway的内存压力会增大，需要谨慎使用）

val numberArray = Array(1, 2, 3, 4, 5, 6, 7, 8, 9, 10)
        val numbers = sc.parallelize(numberArray, 1)  
        val top3Numbers = numbers.take(3)
        for(num <- top3Numbers) {
          println(num)  
        }
    //1,2,3

**first()** : 返回数据集的第一个元素（类似于take(1)）   
**saveAsTextFile(path)** : 将数据集的元素，以textfile的形式，保存到本地文件系统，hdfs或者任何其它hadoop支持的文件系统。Spark将会调用每个元素的toString方法，并将它转换为文件中的一行文本   
**saveAsSequenceFile(path)** : 将数据集的元素，以sequencefile的格式，保存到指定的目录下，本地系统，hdfs或者任何其它hadoop支持的文件系统。RDD的元素必须由key-value对组成，并都实现了Hadoop的Writable接口，或隐式可以转换为**Writable**（Spark包括了基本类型的转换，例如Int，Double，String等等）   
**foreach(func)** : 在数据集的每一个元素上，运行函数func。这通常用于更新一个累加器变量，或者和外部存储系统做交互

### 共享变量示例  ###

Spark提供的Broadcast Variable是只读的，可以通过调用SparkContext的broadcast()方法来针对某个变量创建广播变量。每个节点可以使用广播变量的value()方法来获取值。

val factor = 3
       val factorBroadcast = sc.broadcast(factor)
    
       val arr =Array(1,2,3,4,5)
       val rdd = sc.parallelize(arr)
       val mutiRdd = rdd.map(num => num*factorBroadcast.value)
       mutiRdd.foreach(num => println(num))
    //结果
    3
    6
    9
    12
    15

Spark提供的Accumulator，主要用于多个节点对一个变量进行共享的操作，task只能对Accumulator进行累加操作，不能读取它的值，只有Driver程序可以读取

val sumAccumulator = sc.accumulator(0)
    
        val arr = Array(1,2,3,4,5)
        val rdd = sc.parallelize(arr)
        rdd.foreach(num => sumAccumulator +=num)
        println(sumAccumulator.value)
    //结果 15

[image_1arhvvo24sjq5921hk81ile1ob19.png]: /images/20210923/ee5236b089944e2798e5feaddea5dae3.png
[image_1ari04fp91q8sdfb1b8r66tj4mm.png]: /images/20210923/68b604d5c43945eaa69bd0f8c900e738.png
[image_1ari0mqic1qmq1qvl1q2f1sq91dl13.png]: /images/20210923/a0494d4380bb4373849b98724f361a89.png
[image_1ari0pnrr31b8pkgn141f1p661g.png]: /images/20210923/822326fdf3ca4494854cdf96c5cebec3.png
[image_1ari0vb9ie2f1sb7ehc13tvl411t.png]: /images/20210923/cbe655f6f67849229ee5817d86e6ecb1.png
[image_1ari12cb8122f11tv1ufv14ao1i4n2a.png]: /images/20210923/bf568061d67a43bf919854a5ae7d5b1c.png