spark的RDD、算子、持久化算子分别是什么

本篇内容主要讲解“spark的RDD、算子、持久化算子分别是什么”，感兴趣的朋友不妨来看看。本文介绍的方法操作简单快捷，实用性强。下面就让小编来带大家学习“spark的RDD、算子、持久化算子分别是什么”吧!

一：RDD的介绍

RDD(Resilient Distributed Dateset)，弹性分布式数据集。
RDD的五大特性：
1.RDD是由一系列的partition组成的。
2.函数是作用在每一个partition（split）上的。
3.RDD之间有一系列的依赖关系。
4.分区器是作用在K,V格式的RDD上。
5.RDD提供一系列最佳的计算位置。

1. 注意：

• textFile方法底层封装的是读取MR读取文件的方式，读取文件之前先split，默认split大小是一个block大小。

• RDD实际上不存储数据，这里方便理解，暂时理解为存储数据。

• 什么是K,V格式的RDD?

• 如果RDD里面存储的数据都是二元组对象，那么这个RDD我们就叫做K,V格式的RDD。

• 哪里体现RDD的弹性（容错）？

• partition数量，大小没有限制，体现了RDD的弹性。

• RDD之间依赖关系，可以基于上一个RDD重新计算出RDD。

• 哪里体现RDD的分布式？

• RDD是由Partition组成，partition是分布在不同节点上的。RDD提供计算最佳位置，体现了数据本地化。体现了大数据中“计算移动数据不移动”的理念。

RDD的创建方式

XX.parallelize()
XX.makeRDD()

二、算子

Transformations转换算子有如下：

filter
过滤符合条件的记录数，true保留，false过滤掉

map
将一个RDD中的每个数据项，通过map中的函数映射变为一个新的元素
特点：输入一条，输出一条数据

flatMap
先map后flat。与map类似，每个输入项可以映射为0到多个输出项

sample
随机抽样算子，根据传进去的小数按比例进行又放回或者无放回的抽样

reduceByKey
将相同的Key根据相应的逻辑进行处理

sortByKey/sortBy
作用在K,V格式的RDD上，对key进行升序或者降序排序

join,leftOuterJoin,rightOuterJoin,fullOuterJoin
作用在K,V格式的RDD上。根据K进行连接，对（K,V）join(K,W)返回（K,(V,W)）,join后的分区数与父RDD分区数多的那一个相同。

union
合并两个数据集。两个数据集的类型要一致,返回新的RDD的分区数是合并RDD分区数的总和。

intersection
取两个数据集的交集，返回新的RDD与父RDD分区多的一致

subtract
取两个数据集的差集，结果RDD的分区数与subtract前面的RDD的分区数一致。

mapPartitions
与map类似，遍历的单位是每个partition上的数据。

distinct
实际内部使用的是map+reduceByKey+map，就是去重的意思

cogroup 
当调用类型（K,V）和（K，W）的数据上时，返回一个数据集（K，（Iterable<V>,Iterable<W>）），子RDD的分区与父RDD多的一致

mapPartitionWithIndex
类似于mapPartitions,除此之外还会携带分区的索引值。

repartition
增加或减少分区。会产生shuffle。（多个分区分到一个分区不会产生shuffle）

coalesce
用来减少分区，第二个参数是减少分区的过程中是否产生shuffle。true为产生shuffle，false不产生shuffle。默认是false。
如果coalesce设置的分区数比原来的RDD的分区数还多的话，第二个参数设置为false不会起作用，如果设置成true，效果和repartition一样。即repartition(numPartitions) = coalesce(numPartitions,true)

groupByKey
作用在K，V格式的RDD上。根据Key进行分组。作用在（K，V），返回（K，Iterable <V>）。

zip
将两个RDD中的元素（KV格式/非KV格式）变成一个KV格式的RDD,两个RDD的每个分区元素个数必须相同

zipWithIndex
该函数将RDD中的元素和这个元素在RDD中的索引号（从0开始）组合成（K,V）对。

Action行动算子有如下：

count
返回数据集中的元素数。会在结果计算完成后回收到Driver端

take(n)
返回一个包含数据集前n个元素的集合

first
first=take(1),返回数据集中的第一个元素

foreach
循环遍历数据集中的每个元素，运行相应的逻辑

collect
将计算结果回收到Driver端

转换算子和行动算子的区别：

大多数情况下转换算子返回的内心是RDD类型，行动算子返回的类型大多常常是普通类型，当不知道算子是属于那个类型算子的时候，可以用这个推测

最后一种叫做：控制算子，分别是cache、persist、checkpoint,cache和persist都是懒执行的。必须有一个action类算子触发执行。checkpoint算子不仅能将RDD持久化到磁盘，还能切断RDD之间的依赖关系

三、案列

案列1：cache

请求数据：words数据有约1700万条记录，文件大小约200m大小

结果：

 没用做持久化算子处理：56995 毫秒
 cache: 274 毫秒

scala代码

 def main(args: Array[String]): Unit = {
    /*cache*/
   val conf = new SparkConf()
    conf.setAppName("spark04")
    conf.setMaster("local")

    val context = new SparkContext(conf)
    val sc = context.textFile("./words")
    val result = sc.flatMap(_.split("")).map((_,1)).reduceByKey((_+_))

    val startTime1 = System.currentTimeMillis()
    //默认将RDD的数据持久化到内存中。cache是懒执行,使用persist时候，与cache搭配使用，效率要高点
    //不能这样写：rdd.cache().count()返回的不是持久化的RDD，而是一个数值了
    result.count()
    println("没用做持久化算子处理："+(System.currentTimeMillis()-startTime1)+" 毫秒")

    val startTime2 = System.currentTimeMillis()
    //默认将RDD的数据持久化到内存中。cache是懒执行
    result.cache()
    //cache和persist都是懒执行，必须有一个action类算子触发执行
    result.count()
    println("cache: "+(System.currentTimeMillis()-startTime2).toString+" 毫秒")
  }

案列2: persist

数据：words数据有约1700万条记录，文件大小约200m大小

结果：

没用做持久化算子处理：55350毫秒 persist: 312 毫秒

scala代码

 def main(args: Array[String]): Unit = {
    /*cache*/
   val conf = new SparkConf()
    conf.setAppName("spark04")
    conf.setMaster("local")

    val context = new SparkContext(conf)
    val sc = context.textFile("./words")
    val result = sc.flatMap(_.split("")).map((_,1)).reduceByKey((_+_))

    val startTime1 = System.currentTimeMillis()
    result.count()
    println("没用做持久化算子处理："+(System.currentTimeMillis()-startTime1)+" 毫秒")

    val startTime2 = System.currentTimeMillis()
    //默认将RDD的数据持久化到内存中。cache是懒执行,使用persist时候，与cache搭配使用，效率要高点
    //不能这样写：rdd.persist().count()返回的不是持久化的RDD，而是一个数值了
    //persist()内部就是调用cache()
    result.persist()
    //1cache和persist都是懒执行，必须有一个action类算子触发执行。
    result.count()
    println("persist: "+(System.currentTimeMillis()-startTime2).toString+" 毫秒")
  }
}

案列3： checkpoint

数据：words数据有约1700万条记录，文件大小约200m大小

结果：

没用做持久化算子处理：55575 毫秒
checkpoint: 55851 毫秒
同时会在指定的输出路径中多出持久化到硬盘的数据文件

由于要持久化到硬盘中，速度要慢很多
可以与result.cache()搭配使用

搭配使用后的运行结果
checkpoint: 54621 毫秒
比之前的少了1000多毫秒，result使用cache后，以前处理的数据都放到缓存中去了，所以要稍微快一点

scala代码

 def main(args: Array[String]): Unit = {
    /*cache*/
   val conf = new SparkConf()
    conf.setAppName("spark04")
    conf.setMaster("local")

    val context = new SparkContext(conf)
    //设置checkpoint输出路径
    context.setCheckpointDir("./checkpoint_file")
    //数据来源
    val sc = context.textFile("./words")

    val result = sc.flatMap(_.split("")).map((_,1)).reduceByKey((_+_))

    val startTime1 = System.currentTimeMillis()
    result.count()
    println("没用做持久化算子处理："+(System.currentTimeMillis()-startTime1)+" 毫秒")

    val startTime2 = System.currentTimeMillis()
    //不能这样：rdd.checkpoint().count()
    result.checkpoint()
    /** 一个action类原子就是会启动一个job
      1.当RDD的job执行完毕后，会从finalRDD从后往前回溯。
      2.当回溯到某一个RDD调用了checkpoint方法，会对当前的RDD做一个标记。
      3.Spark框架会自动启动一个新的job，重新计算这个RDD的数据，将数据持久化到HDFS上。
      优化：对RDD执行checkpoint之前，最好对这个RDD先执行cache，这样新启动的job只需要将内存中的数据拷贝到HDFS上就可以，省去了重新计算这一步
      */
    //result.cache()
    result.count()
    println("checkpoint: "+(System.currentTimeMillis()-startTime2).toString+" 毫秒")
  }

总结：

cache和persist的注意事项：
1.cache和persist都是懒执行，必须有一个action类算子触发执行。
2.cache和persist算子的返回值可以赋值给一个变量，在其他job中直接使用这个变量就是使用持久化的数据了。持久化的单位是partition。
3.cache和persist算子后不能立即紧跟action算子。
4.cache和persist算子持久化的数据当applilcation执行完成之后会被清除。
错误：rdd.cache().count() 返回的不是持久化的RDD，而是一个数值了。

checkpoint
checkpoint将RDD持久化到磁盘，还可以切断RDD之间的依赖关系。checkpoint目录数据当application执行完之后不会被清除。
checkpoint 的执行原理：
1.当RDD的job执行完毕后，会从finalRDD从后往前回溯。
2.当回溯到某一个RDD调用了checkpoint方法，会对当前的RDD做一个标记。
3.Spark框架会自动启动一个新的job，重新计算这个RDD的数据，将数据持久化到HDFS上。
优化：对RDD执行checkpoint之前，最好对这个RDD先执行cache，这样新启动的job只需要将内存中的数据拷贝到HDFS上就可以，省去了重新计算这一步

到此，相信大家对“spark的RDD、算子、持久化算子分别是什么”有了更深的了解，不妨来实际操作一番吧！这里是亿速云网站，更多相关内容可以进入相关频道进行查询，关注我们，继续学习！