开发: C++知识库 Java知识库 JavaScript Python PHP知识库人工智能区块链大数据移动开发嵌入式开发工具数据结构与算法开发测试游戏开发网络协议系统运维
教程: HTML教程 CSS教程 JavaScript教程 Go语言教程 JQuery教程 VUE教程 VUE3教程 Bootstrap教程 SQL数据库教程 C语言教程 C++教程 Java教程 Python教程 Python3教程 C#教程
数码: 电脑笔记本显卡显示器固态硬盘硬盘耳机手机 iphone vivo oppo 小米华为单反装机图拉丁

-> 大数据 -> 大数据Spark RDD 函数 -> 正文阅读

[大数据]大数据Spark RDD 函数

1 函数分类

有一定开发经验的读者应该都使用过多线程，利用多核 CPU 的并行能力来加快运算速率。在开发并行程序时，可以利用类似 Fork/Join 的框架将一个大的任务切分成细小的任务，每个小任务模块之间是相互独立的，可以并行执行，然后将所有小任务的结果汇总起来，得到最终的结果。一个非常好的例子便是归并排序。对整个序列进行排序时，可以将序列切分成多个子序列进行排序，然后将排好序的子序列归并起来得到最终的结果。
在这里插入图片描述
对 Hadoop 有所了解的读者都知道 map、reduce 操作。对于大量的数据，我们可以通过map 操作让不同的集群节点并行计算，之后通过 reduce 操作将结果整合起来得到最终输出。

对于 Spark 处理的大量数据而言，会将数据切分后放入RDD作为Spark 的基本数据结构，开发者可以在 RDD 上进行丰富的操作，之后 Spark 会根据操作调度集群资源进行计算。总结起来，RDD 的操作主要可以分为 Transformation 和 Action 两种。
在这里插入图片描述
官方文档：http://spark.apache.org/docs/latest/rdd-programming-guide.html#rdd-operations
RDD中操作（函数、算子）分为两类：

1）、Transformation转换操作：返回一个新的RDD

which create a new dataset from an existing one
所有Transformation函数都是Lazy，不会立即执行，需要Action函数触发

2）、Action动作操作：返回值不是RDD(无返回值或返回其他的)

which return a value to the driver program after running a computation on the datase
所有Action函数立即执行（Eager），比如count、first、collect、take等

此外注意RDD中函数细节：

第一点：RDD不实际存储真正要计算的数据，而是记录了数据的位置在哪里，数据的转换关系(调用了什么方法，传入什么函数)；
第二点：RDD中的所有转换都是惰性求值/延迟执行的，也就是说并不会直接计算。只有当发生一个要求返回结果给Driver的Action动作时，这些转换才会真正运行。之所以使用惰性求值/
延迟执行，是因为这样可以在Action时对RDD操作形成DAG有向无环图进行Stage的划分和并行优化，这种设计让Spark更加有效率地运行。

2 Transformation函数

在Spark中Transformation操作表示将一个RDD通过一系列操作变为另一个RDD的过程，这个操作可能是简单的加减操作，也可能是某个函数或某一系列函数。值得注意的是Transformation操作并不会触发真正的计算，只会建立RDD间的关系图。
如下图所示，RDD内部每个方框是一个分区。假设需要采样50%的数据，通过sample函数，从 V1、V2、U1、U2、U3、U4 采样出数据 V1、U1 和 U4，形成新的RDD。在这里插入图片描述
常用Transformation转换函数，加上底色为重要函数，重点讲解常使用函数：

3 Action函数

不同于Transformation操作，Action操作代表一次计算的结束，不再产生新的 RDD，将结果返回到Driver程序或者输出到外部。所以Transformation操作只是建立计算关系，而Action 操作才是实际的执行者。每个Action操作都会调用SparkContext的runJob 方法向集群正式提交请求，所以每个Action操作对应一个Job。
常用Action执行函数，加上底色为重要函数，后续重点讲解。
在这里插入图片描述

4 重要函数

RDD中包含很多函数，主要可以分为两类：Transformation转换函数和Action函数。在这里插入图片描述
主要常见使用函数如下，一一通过演示范例讲解。

4.1 基本函数

RDD中map、filter、flatMap及foreach等函数为最基本函数，都是都RDD中每个元素进行操作，将元素传递到函数中进行转换。

map 函数：

map(f:T=>U) : RDD[T]=>RDD[U]，表示将 RDD 经由某一函数 f 后，转变为另一个RDD。

flatMap 函数：

flatMap(f:T=>Seq[U]) : RDD[T]=>RDD[U])，表示将 RDD 经由某一函数 f 后，转变为一
个新的 RDD，但是与 map 不同，RDD 中的每一个元素会被映射成新的 0 到多个元素
（f 函数返回的是一个序列 Seq）。

filter 函数：

filter(f:T=>Bool) : RDD[T]=>RDD[T]，表示将 RDD 经由某一函数 f 后，只保留 f 返回
为 true 的数据，组成新的 RDD。

foreach 函数：

foreach(func)，将函数 func 应用在数据集的每一个元素上，通常用于更新一个累加器，或者和外部存储系统进行交互，例如 Redis。关于 foreach，在后续章节中还会使用，到时会详细介绍它的使用方法及注意事项。

saveAsTextFile 函数：

saveAsTextFile(path:String)，数据集内部的元素会调用其 toString 方法，转换为字符串形式，然后根据传入的路径保存成文本文件，既可以是本地文件系统，也可以是HDFS 等。上述函数基本上都使用过，在后续的案例中继续使用，此处不再单独演示案例。

4.2 分区操作函数

每个RDD由多分区组成的，实际开发建议对每个分区数据的进行操作，map函数使用mapPartitions代替、foreache函数使用foreachPartition代替。
在这里插入图片描述
针对词频统计WordCount代码进行修改，针对分区数据操作，范例代码如下

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

/**
 * 分区操作函数：mapPartitions和foreachPartition
 */
object SparkIterTest {
  def main(args: Array[String]): Unit = {
    // 创建应用程序入口SparkContext实例对象
    val sc: SparkContext = {
      // 1.a 创建SparkConf对象，设置应用的配置信息
      val sparkConf: SparkConf = new SparkConf()
        .setAppName(this.getClass.getSimpleName.stripSuffix("$"))
        .setMaster("local[2]")
      // 1.b 传递SparkConf对象，构建Context实例
      new SparkContext(sparkConf)
    }
    sc.setLogLevel("WARN")
    // TODO: 1、从文件系统加载数据，创建RDD数据集
    val inputRDD: RDD[String] = sc.textFile("datas/wordcount/wordcount.data", minPartitions = 2)
    // TODO: 2、处理数据，调用RDD集合中函数（类比于Scala集合类中列表List）
    /*
    def mapPartitions[U: ClassTag](
    f: Iterator[T] => Iterator[U],
    preservesPartitioning: Boolean = false
    ): RDD[U]
    */
    val wordcountsRDD: RDD[(String, Int)] = inputRDD
      // 将每行数据按照分隔符进行分割，将数据扁平化
      .flatMap(line => line.trim.split("\\s+"))
      // TODO: 针对每个分区数据操作
      .mapPartitions { iter =>
        // iter 表示RDD中每个分区中的数据，存储在迭代器中，相当于列表List
        iter.map(word => (word, 1))
      }
      // 按照Key聚合统计, 先按照Key分组，再聚合统计（此函数局部聚合，再进行全局聚合）
      .reduceByKey((a, b) => a + b)
    // TODO: 3、输出结果RDD到本地文件系统
    wordcountsRDD.foreachPartition { datas =>
      // 获取各个分区ID
      val partitionId: Int = TaskContext.getPartitionId()
      // val xx: Iterator[(String, Int)] = datas
      datas.foreach { case (word, count) =>
        println(s"p-${partitionId}: word = $word, count = $count")
      }
    }
    // 应用程序运行结束，关闭资源
    sc.stop()
  }
}

为什么要对分区操作，而不是对每个数据操作，好处在哪里呢？？？

应用场景：处理网站日志数据，数据量为10GB，统计各个省份PV和UV。

假设10GB日志数据，从HDFS上读取的，此时RDD的分区数目：80 分区；
但是分析PV和UV有多少条数据：34，存储在80个分区中，实际项目中降低分区数目，比
如设置为2个分区。
.

4.3 重分区函数

如何对RDD中分区数目进行调整（增加分区或减少分区），在RDD函数中主要有如下三个函数。

1）、增加分区函数

函数名称：repartition，此函数使用的谨慎，会产生Shuffle。

2）、减少分区函数

函数名称：coalesce，此函数不会产生Shuffle，当且仅当降低RDD分区数目。
比如RDD的分区数目为10个分区，此时调用rdd.coalesce(12)，不会对RDD进行任何操作。

3）、调整分区函数

在PairRDDFunctions（此类专门针对RDD中数据类型为KeyValue对提供函数）工具类中
partitionBy函数：

范例演示代码，适当使用函数调整RDD分区数目：

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

/**
 * RDD中分区函数，调整RDD分区数目，可以增加分区和减少分区
 */
object SparkPartitionTest {
  def main(args: Array[String]): Unit = {
    // 创建应用程序入口SparkContext实例对象
    val sc: SparkContext = {
      // 1.a 创建SparkConf对象，设置应用的配置信息
      val sparkConf: SparkConf = new SparkConf()
        .setAppName(this.getClass.getSimpleName.stripSuffix("$"))
        .setMaster("local[2]")
      // 1.b 传递SparkConf对象，构建Context实例
      new SparkContext(sparkConf)
    }
    sc.setLogLevel("WARN")
    // 读取本地文件系统文本文件数据
    val datasRDD: RDD[String] = sc.textFile("datas/wordcount/wordcount.data", minPartitions = 2)
    // TODO: 增加RDD分区数
    val etlRDD: RDD[String] = datasRDD.repartition(3)
    println(s"EtlRDD 分区数目 = ${etlRDD.getNumPartitions}")
    // 词频统计
    val resultRDD: RDD[(String, Int)] = etlRDD
      // 数据分析，考虑过滤脏数据
      .filter(line => null != line && line.trim.length > 0)
      // 分割单词，注意去除左右空格
      .flatMap(line => line.trim.split("\\s+"))
      // 转换为二元组，表示单词出现一次
      .mapPartitions { iter =>
        iter.map(word => (word, 1))
      }
      // 分组聚合，按照Key单词
      .reduceByKey((tmp, item) => tmp + item)
    // 输出结果RDD
    resultRDD
      // TODO: 对结果RDD降低分区数目
      .coalesce(1)
      .foreachPartition(iter => iter.foreach(println))
    // 应用程序运行结束，关闭资源
    sc.stop()
  }
}

在实际开发中，什么时候适当调整RDD的分区数目呢？让程序性能更好好呢？？？？

第一点：增加分区数目

当处理的数据很多的时候，可以考虑增加RDD的分区数目

第二点：减少分区数目

其一：当对RDD数据进行过滤操作（filter函数）后，考虑是否降低RDD分区数目
其二：当对结果RDD存储到外部系统

4.4 聚合函数

在数据分析领域中，对数据聚合操作是最为关键的，在Spark框架中各个模块使用时，主要就
是其中聚合函数的使用。

4.4.1 集合中聚合函数

回顾列表List中reduce聚合函数核心概念：聚合的时候，往往需要聚合中间临时变量。查看列
表List中聚合函数reduce和fold源码如下：在这里插入图片描述

通过代码，看看列表List中聚合函数使用：
在这里插入图片描述
运行截图如下所示：

fold聚合函数，比reduce聚合函数，多提供一个可以初始化聚合中间临时变量的值参数：

聚合操作时，往往聚合过程中需要中间临时变量（到底时几个变量，具体业务而定），如下案例：在这里插入图片描述

4.4.2 RDD 中聚合函数

在RDD中提供类似列表List中聚合函数reduce和fold，查看如下：在这里插入图片描述
案例演示：求列表List中元素之和，RDD中分区数目为2，核心业务代码如下：
运行原理分析：
使用RDD中fold聚合函数：

查看RDD中高级聚合函数aggregate，函数声明如下：
业务需求：使用aggregate函数实现RDD中最大的两个数据，分析如下：在这里插入图片描述
核心业务代码如下：

运行结果原理剖析示意图：
上述完整范例演示代码：

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext, TaskContext}
import scala.collection.mutable
import scala.collection.mutable.ListBuffer

/**
 * RDD中聚合函数：reduce、aggregate函数
 */
object SparkAggTest {
  def main(args: Array[String]): Unit = {
    // 创建应用程序入口SparkContext实例对象
    val sc: SparkContext = {
      // 1.a 创建SparkConf对象，设置应用的配置信息
      val sparkConf: SparkConf = new SparkConf()
        .setAppName(this.getClass.getSimpleName.stripSuffix("$"))
        .setMaster("local[2]")
      // 1.b 传递SparkConf对象，构建Context实例
      new SparkContext(sparkConf)
    }
    sc.setLogLevel("WARN")
    // 模拟数据，1 到 10 的列表，通过并行方式创建RDD
    val datas = 1 to 10
    val datasRDD: RDD[Int] = sc.parallelize(datas, numSlices = 2)
    // 查看每个分区中的数据
    datasRDD.foreachPartition { iter =>
      println(s"p-${TaskContext.getPartitionId()}: ${iter.mkString(", ")}")
    }
    println("=========================================")
    // 使用reduce函数聚合
    val result: Int = datasRDD.reduce((tmp, item) => {
      println(s"p-${TaskContext.getPartitionId()}: tmp = $tmp, item = $item")
      tmp + item
    })
    println(result)
    println("=========================================")
    // 使用fold函数聚合
    val result2: Int = datasRDD.fold(0)((tmp, item) => {
      println(s"p-${TaskContext.getPartitionId()}: tmp = $tmp, item = $item")
      tmp + item
    })
    println(result2)
    println("=========================================")
    // 使用aggregate函数获取最大的两个值
    val top2: mutable.Seq[Int] = datasRDD.aggregate(new ListBuffer[Int]())(
      // 分区内聚合函数，每个分区内数据如何聚合 seqOp: (U, T) => U,
      (u, t) => {
        println(s"p-${TaskContext.getPartitionId()}: u = $u, t = $t")
        // 将元素加入到列表中
        u += t //
        // 降序排序
        val top = u.sorted.takeRight(2)
        // 返回
        top
      },
      // 分区间聚合函数，每个分区聚合的结果如何聚合 combOp: (U, U) => U
      (u1, u2) => {
        println(s"p-${TaskContext.getPartitionId()}: u1 = $u1, u2 = $u2")
        u1 ++= u2 // 将列表的数据合并，到u1中
        //
        u1.sorted.takeRight(2)
      }
    )
    println(top2)
    // 应用程序运行结束，关闭资源
    sc.stop()
  }
}

4.4.3 PairRDDFunctions 聚合函数

在Spark中有一个object对象PairRDDFunctions，主要针对RDD的数据类型是Key/Value对的数
据提供函数，方便数据分析处理。比如使用过的函数：reduceByKey、groupByKey等。*ByKey函
数：将相同Key的Value进行聚合操作的，省去先分组再聚合。

第一类：分组函数groupByKey
第二类：分组聚合函数reduceByKey和foldByKey

但是reduceByKey和foldByKey聚合以后的结果数据类型与RDD中Value的数据类型是一样的。
第三类：分组聚合函数aggregateByKey
在企业中如果对数据聚合使用，不能使用reduceByKey完成时，考虑使用aggregateByKey函数，
基本上都能完成任意聚合功能。
演示范例代码如下：

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

/**
 * RDD中聚合函数，针对RDD中数据类型Key/Value对：
 * groupByKey
 * reduceByKey/foldByKey
 * aggregateByKey
 * combineByKey
 */
object SparkAggByKeyTest {
  def main(args: Array[String]): Unit = {
    // 创建应用程序入口SparkContext实例对象
    val sc: SparkContext = {
      // 1.a 创建SparkConf对象，设置应用的配置信息
      val sparkConf: SparkConf = new SparkConf()
        .setAppName(this.getClass.getSimpleName.stripSuffix("$"))
        .setMaster("local[2]")
      // 1.b 传递SparkConf对象，构建Context实例
      new SparkContext(sparkConf)
    }
    sc.setLogLevel("WARN")
    // 1、并行化集合创建RDD数据集
    val linesSeq: Seq[String] = Seq(
      "hadoop scala hive spark scala sql sql", //
      "hadoop scala spark hdfs hive spark", //
      "spark hdfs spark hdfs scala hive spark" //
    )
    val inputRDD: RDD[String] = sc.parallelize(linesSeq, numSlices = 2)
    // 2、分割单词，转换为二元组
    val wordsRDD: RDD[(String, Int)] = inputRDD
      .flatMap(line => line.split("\\s+"))
      .map(word => word -> 1)
    // TODO: 先使用groupByKey函数分组，再使用map函数聚合
    val wordsGroupRDD: RDD[(String, Iterable[Int])] = wordsRDD.groupByKey()
    val resultRDD: RDD[(String, Int)] = wordsGroupRDD.map { case (word, values) =>
      val count: Int = values.sum
      word -> count
    }
    println(resultRDD.collectAsMap())
    // TODO: 直接使用reduceByKey或foldByKey分组聚合
    val resultRDD2: RDD[(String, Int)] = wordsRDD.reduceByKey((tmp, item) => tmp + item)
    println(resultRDD2.collectAsMap())
    val resultRDD3 = wordsRDD.foldByKey(0)((tmp, item) => tmp + item)
    println(resultRDD3.collectAsMap())
    // TODO: 使用aggregateByKey聚合
    /*
    def aggregateByKey[U: ClassTag]
    (zeroValue: U) // 聚合中间临时变量初始值，类似fold函数zeroValue
    (
    seqOp: (U, V) => U, // 各个分区内数据聚合操作函数
    combOp: (U, U) => U // 分区间聚合结果的聚合操作函数
    ): RDD[(K, U)]
    */
    val resultRDD4 = wordsRDD.aggregateByKey(0)(
      (tmp: Int, item: Int) => {
        tmp + item
      },
      (tmp: Int, result: Int) => {
        tmp + result
      }
    )
    println(resultRDD4.collectAsMap())
    // 应用程序运行结束，关闭资源
    Thread.sleep(1000000)
    sc.stop()
  }
}

4.4.4 面试题

RDD中groupByKey和reduceByKey区别？？？

reduceByKey函数：在一个(K,V)的RDD上调用，返回一个(K,V)的RDD，使用指定的reduce函数，
将相同key的值聚合到一起，reduce任务的个数可以通过第二个可选的参数来设置。
groupByKey函数：在一个(K,V)的RDD上调用，返回一个(K,V)的RDD，使用指定的函数，将相同
key的值聚合到一起，与reduceByKey的区别是只生成一个sequence。

4.5 关联函数

当两个RDD的数据类型为二元组Key/Value对时，可以依据Key进行关联Join。在这里插入图片描述
首先回顾一下SQL JOIN，用Venn图表示如下：
RDD中关联JOIN函数都在PairRDDFunctions中，具体截图如下：
具体看一下join（等值连接）函数说明：
范例演示代码：

import org.apache.sp
import org.apache.spark.{SparkConf, SparkContext}

/**
 * RDD中关联函数Join，针对RDD中数据类型为Key/Value对
 *
 */
object SparkJoinTest {
  def main(args: Array[String]): Unit = {
    // 创建应用程序入口SparkContext实例对象
    val sc: SparkContext = {
      // 1.a 创建SparkConf对象，设置应用的配置信息
      val sparkConf: SparkConf = new SparkConf()
        .setAppName(this.getClass.getSimpleName.stripSuffix("$"))
        .setMaster("local[2]")
      // 1.b 传递SparkConf对象，构建Context实例
      new SparkContext(sparkConf)
    }
    sc.setLogLevel("WARN")
    // 模拟数据集
    val empRDD: RDD[(Int, String)] = sc.parallelize(
      Seq((1001, "zhangsan"), (1002, "lisi"), (1003, "wangwu"), (1004, "zhangliu"))
    )
    val deptRDD: RDD[(Int, String)] = sc.parallelize(
      Seq((1001, "sales"), (1002, "tech"))
    )
    /*
    def join[W](other: RDD[(K, W)]): RDD[(K, (V, W))]
    */
    val joinRDD: RDD[(Int, (String, String))] = empRDD.join(deptRDD)
    println(joinRDD.collectAsMap())
    /*
    def leftOuterJoin[W](other: RDD[(K, W)]): RDD[(K, (V, Option[W]))]
    */
    val leftJoinRDD: RDD[(Int, (String, Option[String]))] = empRDD.leftOuterJoin(deptRDD)
    println(leftJoinRDD.collectAsMap())
    // 应用程序运行结束，关闭资源
    sc.stop()
  }
}