bigdata-37-SparkRDD

简介: bigdata-37-SparkRDD

RDD持久化原理

Spark中有一个非常重要的功能就是可以对RDD进行持久化。当对RDD执行持久化操作时,每个节点都会将自己操作的RDD的partition数据持久化到内存中,并且在之后对该RDD的反复使用中,直接使用内存中缓存的partition数据。

这样的话,针对一个RDD反复执行多个操作的场景,就只需要对RDD计算一次即可,后面直接使用该RDD,而不需要反复计算多次该RDD。

因为正常情况下这个RDD的数据使用过后内存中是不会一直保存的。

例如这样的操作:针对mapRDD需要多次使用的

val dataRDD = sc.parallelize(Array(1,2,3,4,5))

val mapRDD = dataRDD.map(...)

mapRDD.foreach(...)

mapRDD.saveAsTextFile(...)

mapRDD.collect()

巧妙使用RDD持久化,在某些场景下,对spark应用程序的性能有很大提升。

特别是对于迭代式算法和快速交互式应用来说,RDD持久化,是非常重要的。

要持久化一个RDD,只需要调用它的cache()或者persist()方法就可以了。

在该RDD第一次被计算出来时,就会直接缓存在每个节点中。而且Spark的持久化机制还是自动容错的,如果持久化的RDD的任何partition数据丢失了,那么Spark会自动通过其源RDD,使用transformation算子重新计算该partition的数据。

cache()和persist()的区别在于:

cache()是persist()的一种简化方式,cache()的底层就是调用的persist()的无参版本,也就是调用persist(MEMORY_ONLY),将数据持久化到内存中。

如果需要从内存中清除缓存,那么可以使用unpersist()方法。

RDD持久化策略

下面看一下目前Spark支持的一些持久化策略

MEMORY_ONLY:以非序列化的Java对象的方式持久化在JVM内存中。如果内存无法完全存储RDD所有的partition,那么那些没有持久化的partition就会在下一次需要使用它的时候,重新被计算。

MEMORY_AND_DISK:当某些partition无法存储在内存中时,会持久化到磁盘中。下次需要使用这些partition时,需要从磁盘上读取,不需要重新计算

MEMORY_ONLY_SER:同MEMORY_ONLY,但是会使用Java的序列化方式,将Java对象序列化后进行持久化。可以减少内存开销,但是在使用的时候需要进行反序列化,因此会增加CPU开销。

MEMORY_AND_DISK_SER:同MEMORY_AND_DSK。但是会使用序列化方式持久化Java对象。

DISK_ONLY:使用非序列化Java对象的方式持久化,完全存储到磁盘上。

MEMORY_ONLY_2、MEMORY_AND_DISK_2等:如果是尾部加了2的持久化级别,表示会将持久化数据复制一份,保存到其它节点,从而在数据丢失时,不需要重新计算,只需要使用备份数据即可。

如何选择RDD持久化策略

Spark提供了多种持久化级别,主要是为了在CPU和内存消耗之间进行取舍。

下面是一些通用的持久化级别的选择建议:

  1. 优先使用MEMORY_ONLY,纯内存速度最快,而且没有序列化不需要消耗CPU进行反序列化操作,缺点就是比较耗内存
  2. MEMORY_ONLY_SER,将数据进行序列化存储,纯内存操作还是非常快,只是在使用的时候需要消耗CPU进行反序列化

如果需要进行数据的快速失败恢复,那么就选择带后缀为_2的策略,进行数据的备份,这样在失败时,就不需要重新计算了能不使用DISK相关的策略,就不要使用,因为有的时候,从磁盘读取数据,还不如重新计算一次。

案例:Scala实现RDD持久化

package com.bigdata.scala

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

/**

* 需求:RDD持久化

*/

object PersistRddScala {

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

   val conf = new SparkConf()

   conf.setAppName("PersistRddScala")

     .setMaster("local")

   val sc = new SparkContext(conf)

   //注意cache的用法和位置

   //cache默认是基于内存的持久化

   // cache()=persist()=persist(StorageLevel.MEMORY_ONLY)

   val dataRDD = sc.textFile("/Users/a/keep-learning/test_data/data.txt").cache()

   var start_time = System.currentTimeMillis()

   var count = dataRDD.count()

   println(count)

   var end_time = System.currentTimeMillis()

   println("第一次耗时:"+(end_time-start_time))

   start_time = System.currentTimeMillis()

   count = dataRDD.count()

   println(count)

   end_time = System.currentTimeMillis()

   println("第二次耗时:"+(end_time-start_time))

   sc.stop()

}

}

这里通过对比我们可以看出

在没有添加cache之前,每一次都耗时很长

加上cache之后,第二次计算耗时就很少了

共享变量

共享变量的工作原理

Spark还有一个非常重要的特性就是共享变量

默认情况下,如果在一个算子函数中使用到了某个外部的变量,那么这个变量的值会被拷贝到每个task中。此时每个task只能操作自己的那份变量数据。如果多个task想要共享某个变量,那么这种方式是做不到的。

Spark为此提供了两种共享变量

一种是Broadcast Variable(广播变量)

另一种是Accumulator(累加变量)

Broadcast Variable

Broadcast Variable会将使用到的变量,仅仅为每个节点拷贝一份,而不会为每个task都拷贝一份副本,因此其最大的作用,就是减少变量到各个节点的网络传输消耗,以及在各个节点上的内存消耗

通过调用SparkContext的broadcast()方法,针对某个变量创建广播变量

注意:广播变量,是只读的

然后在算子函数内,使用到广播变量时,每个节点只会拷贝一份副本。可以使用广播变量的value()方法获取值。

从左边分析:

如果是普通的外部变量,算子函数内如果使用到,就会往每个task中拷贝一份,假设这个外部变量是一个集合,集合中有上亿数据,每个task都拷贝一份就会极大的增加网络传输的时间以及内存空间的占用。

从右边分析:

假设使用广播变量,每个变量只会拷贝到一个节点上,存在于每个节点的task变量会共享这一广播变量,从而极大的减少网络传输以及内存占用,提升效率。

基于Scala实现广播变量案例

package com.bigdata.scala

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

object BoradcastOpScala {

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

   val conf = new SparkConf()

   conf.setAppName("BoradcastOpScala")

     .setMaster("local")

   val sc = new SparkContext(conf)

   val dataRDD = sc.parallelize(Array(1,2,3,4,5))

   val varable = 2

  //dataRDD.map(_ * varable)

   //1:定义广播变量

   val varableBroadcast = sc.broadcast(varable)

   //2:使用广播变量,调用其value方法

   dataRDD.map(_ * varableBroadcast.value).foreach(println(_))

   sc.stop()

}

}

结果:

Accumulator

Spark提供的Accumulator,主要用于多个节点对一个变量进行共享性的操作。

正常情况下在Spark的任务中,由于一个算子可能会产生多个task并行执行,所以在这个算子内部执行的聚合计算都是局部的,想要实现多个task进行全局聚合计算,此时需要使用到Accumulator这个共享的累加变量。

注意:Accumulator只提供了累加的功能。在task只能对Accumulator进行累加操作,不能读取它的值。只有在Driver进程中才可以读取Accumulator的值。

package com.bigdata.scala

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

/**

* 需求:使用累加变量

*/

object AccumulatorOpScala {

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

   val conf = new SparkConf()

   conf.setAppName("AccumulatorOpScala")

     .setMaster("local")

   val sc = new SparkContext(conf)

   val dataRDD = sc.parallelize(Array(1,2,3,4,5))

   //这种写法是错误的,因为foreach代码是在worker节点上执行的

   //var total = 0 和 println(total) 是在Driver进程中执行的

   //所以无法实现累加操作

   //并且foreach算子可能会在多个task中执行,这样foreach内部实现的累加也不是最终全局累加的结果

   /*var total = 0

   dataRDD.foreach(num=>total += num)

   println(total)*/

   //所以此时想要实现累加操作就需要使用累加变量了

   //1:定义累加变量

   val sumAccumulator = sc.longAccumulator

   //2:使用累加变量

   dataRDD.foreach(num=>sumAccumulator.add(num))

   //注意:只能在Driver进程中获取累加变量的结果

   println(sumAccumulator.value)

   sc.stop()

}

}

目录
相关文章
|
7月前
|
SQL 分布式计算 资源调度
bigdata-34-Spark初步了解
bigdata-34-Spark初步了解
44 0
|
7月前
|
分布式计算 资源调度 Hadoop
bigdata-06-Hadoop了解与配置
bigdata-06-Hadoop了解与配置
82 0
|
7月前
|
SQL 分布式计算 Hadoop
Hadoop学习笔记(HDP)-Part.16 安装HBase
01 关于HDP 02 核心组件原理 03 资源规划 04 基础环境配置 05 Yum源配置 06 安装OracleJDK 07 安装MySQL 08 部署Ambari集群 09 安装OpenLDAP 10 创建集群 11 安装Kerberos 12 安装HDFS 13 安装Ranger 14 安装YARN+MR 15 安装HIVE 16 安装HBase 17 安装Spark2 18 安装Flink 19 安装Kafka 20 安装Flume
146 1
Hadoop学习笔记(HDP)-Part.16 安装HBase
|
7月前
|
SQL 分布式计算 Hadoop
Hadoop学习笔记(HDP)-Part.15 安装HIVE
01 关于HDP 02 核心组件原理 03 资源规划 04 基础环境配置 05 Yum源配置 06 安装OracleJDK 07 安装MySQL 08 部署Ambari集群 09 安装OpenLDAP 10 创建集群 11 安装Kerberos 12 安装HDFS 13 安装Ranger 14 安装YARN+MR 15 安装HIVE 16 安装HBase 17 安装Spark2 18 安装Flink 19 安装Kafka 20 安装Flume
217 1
Hadoop学习笔记(HDP)-Part.15 安装HIVE
|
7月前
|
分布式计算 资源调度 Java
Hadoop学习笔记(HDP)-Part.17 安装Spark2
01 关于HDP 02 核心组件原理 03 资源规划 04 基础环境配置 05 Yum源配置 06 安装OracleJDK 07 安装MySQL 08 部署Ambari集群 09 安装OpenLDAP 10 创建集群 11 安装Kerberos 12 安装HDFS 13 安装Ranger 14 安装YARN+MR 15 安装HIVE 16 安装HBase 17 安装Spark2 18 安装Flink 19 安装Kafka 20 安装Flume
95 0
Hadoop学习笔记(HDP)-Part.17 安装Spark2
|
7月前
|
存储 机器学习/深度学习 分布式计算
Hadoop学习笔记(HDP)-Part.12 安装HDFS
01 关于HDP 02 核心组件原理 03 资源规划 04 基础环境配置 05 Yum源配置 06 安装OracleJDK 07 安装MySQL 08 部署Ambari集群 09 安装OpenLDAP 10 创建集群 11 安装Kerberos 12 安装HDFS 13 安装Ranger 14 安装YARN+MR 15 安装HIVE 16 安装HBase 17 安装Spark2 18 安装Flink 19 安装Kafka 20 安装Flume
192 0
Hadoop学习笔记(HDP)-Part.12 安装HDFS
|
7月前
|
SQL 消息中间件 分布式计算
Hadoop学习笔记(HDP)-Part.06 安装OracleJDK
01 关于HDP 02 核心组件原理 03 资源规划 04 基础环境配置 05 Yum源配置 06 安装OracleJDK 07 安装MySQL 08 部署Ambari集群 09 安装OpenLDAP 10 创建集群 11 安装Kerberos 12 安装HDFS 13 安装Ranger 14 安装YARN+MR 15 安装HIVE 16 安装HBase 17 安装Spark2 18 安装Flink 19 安装Kafka 20 安装Flume
61 0
Hadoop学习笔记(HDP)-Part.06 安装OracleJDK
|
7月前
|
资源调度 分布式计算 Hadoop
Hadoop学习笔记(HDP)-Part.14 安装YARN+MR
01 关于HDP 02 核心组件原理 03 资源规划 04 基础环境配置 05 Yum源配置 06 安装OracleJDK 07 安装MySQL 08 部署Ambari集群 09 安装OpenLDAP 10 创建集群 11 安装Kerberos 12 安装HDFS 13 安装Ranger 14 安装YARN+MR 15 安装HIVE 16 安装HBase 17 安装Spark2 18 安装Flink 19 安装Kafka 20 安装Flume
280 0
Hadoop学习笔记(HDP)-Part.14 安装YARN+MR
|
7月前
|
SQL 分布式计算 Hadoop
Hadoop学习笔记(HDP)-Part.01 关于HDP
01 关于HDP 02 核心组件原理 03 资源规划 04 基础环境配置 05 Yum源配置 06 安装OracleJDK 07 安装MySQL 08 部署Ambari集群 09 安装OpenLDAP 10 创建集群 11 安装Kerberos 12 安装HDFS 13 安装Ranger 14 安装YARN+MR 15 安装HIVE 16 安装HBase 17 安装Spark2 18 安装Flink 19 安装Kafka 20 安装Flume
223 0
|
XML 存储 分布式计算
Hadoop实战-part4
Hadoop实战-part4
153 0
Hadoop实战-part4