RDD优化：缓存和checkpoint机制、数据共享（广播变量、累加器）、RDD的依赖关系、shuffle过程、并行度说明

1. 缓存和checkpoint机制

缓存和checkpoint也叫作rdd的持久化，将rdd的数据存储在指定位置。
作用：

• 计算容错
• 提升计算速度

1.1 缓存使用

缓存是将数据存储在内存或者磁盘上，缓存的特点是计算结束时，缓存自动清空。

• 缓存级别
  • 指定缓存的数据位置
  • 默认是缓存到内存上
• 使用
  • persist使用该方法
  • cache内部调用persist
  • 手动释放 unpersist

from pyspark import SparkContext
from pyspark.storagelevel import StorageLevelsc = SparkContext()
rdd = sc.parallelize(['a', 'b','a','c'])# rdd数据进行转化
rdd_kv  = rdd.map(lambda x: (x,1))
#rdd_kv数据进行缓存
rdd_kv.persist(storageLevel=StorageLevel.MEMORY_AND_DISK)
#使用action算子触发
rdd_kv.collect()# 分组处理
rdd_group = rdd_kv.groupByKey()#求和计算
rdd_reduce = rdd_group.reduceByKey(lambda x,y:x+y)
# 查看计算结果
res = rdd_reduce.collect()
print(res)

1.2 checkpoint

checkpoint也是将中间rdd数据存储起来，但是存储的位置实时分布式存储系统，可以永久保存，程序结束不会释放。
如果需要删除就在HDFS上删除对应的目录文件。

#checkpoint使用
from pyspark import SparkContext
sc = SparkContext()#使用sc对象指定checkpoint存储位置
sc.setCheckpointDir('hdfs://node1:8020/checkpoint')rdd = sc.parallelize(['a', 'b', 'c', 'd', 'e', 'f', 'g', 'h', 'i', 'j', 'k'])#rdd数据进行转化
rdd_kv = rdd.map(lambda x: (x, 1))#rdd 数据进行checkpoint
rdd_kv.checkpoint()
#需要使用action算子触发checkpoint
print(rdd_kv.glom().collect())#分组处理
rdd_group = rdd_kv.groupByKey()#求和计算
rdd_reduce = rdd_group.reduceByKey(lambda x, y: x + y)#查看计算结果
res = rdd_kv.collect()
print(res)res1 = rdd_group.collect()
print(res1)res2 = rdd_reduce.collect()
print(res2)

1.3 缓存和checkpoint的区别

• 生命周期
  • 缓存数据，程序计算结束后自动删除
  • checkpoint 程序结束，数据依然保留在HDFS
• 存储位置
  • 缓存 优先存储在内存上，也可以选存储在本地磁盘，是在计算任务所在的内存和磁盘上。
  • checkpoint存储在HDFS上
• 依赖关系
  • 缓存数据后，会保留rdd之间依赖关系，缓存临时存储，数据可能会丢失，需要保留依赖，当缓存丢失后可以按照依赖重新计算。
  • checkpoint，数据存储后会断开依赖，数据保存在HDFS，HDFS三副本机制可以保证数据不丢失，所以没有比较保留依赖关系。
    注意：缓存和checkpoint可以作为rdd优化的方案，提升计算速度，一般对经常要使用的rdd进行缓存和checkpoint，对计算比较复杂的rdd进行缓存或checkpoint。

2. 数据共享

2.1 广播变量

如果要在分布式计算里面分发大的变量数据，这个都会由driver端进行分发，一般来讲，如果这个变量不是广播变量，那么每个task都会分发一份，这在task数目十分多的情况下Driver的带宽会成为系统的瓶颈，而且会大量消耗task服务器上的资源，如果将这个变量声明为广播变量，那么每个executor都会拥有一份，这个executor启动的task会共享这个变量，节省了通信的成本和服务器的资源。

减少task线程对应变量的定义，节省内存空间。

#广播变量
from pyspark import SparkContextsc= SparkContext()a_obj = sc.broadcast(10)
#生成rdd
rdd = sc.parallelize([1,2,3,4,5,6,7,8,9])#转化计算
rdd2 = rdd.map(lambda x:x+a_obj.value)#查看数据
res = rdd2.collect()
print(res)

2.2 累加器

避免资源抢占造成错误

#累加器的使用
from pyspark import SparkContextsc = SparkContext()
#将变量值添加到累加器中
acc_obj = sc.accumulator(0)#rdd操作
rdd = sc.parallelize([1,2,3])#使用累加器进行数据累加
rdd2 = rdd.map(lambda x:acc_obj.add(x))#查看结果
res = rdd.collect()
print(res)res1 = rdd2.collect()
print(res1)#查看累加器的结果
print(acc_obj.value)

查看结果：

3. RDD依赖关系

• 窄依赖
  • 每个父RDD的一个partition最多被子RDD的一个partition所使用。
    • map
    • flatMap
    • filter
• 宽依赖
  • 一个父RDD的partition会被多个子RDD的partition所使用
    • groupByKey
    • reduceByKey
    • sortByKey
  • 在宽依赖中rdd之间会发生数据交换，这个交换的过程称为rdd的shuffle
    • 只要是宽依赖必然发生shuffle
    • 在宽依赖进行数据交换时，只有等待所有分区交换完成后，才能进行后续的计算，非常影响计算速度。

那么如何判断是宽依赖还是窄依赖呢？

#判断宽窄依赖
from pyspark import SparkContext
sc = SparkContext()rdd = sc.parallelize([1,2,3,4])
rdd_kv = sc.parallelize([('a',1), ('b',2), ('c',3)])#算子演示
rdd2 = rdd.map(lambda x:x+1)
rdd3 = rdd_kv.groupByKey()#查看结果
# res = rdd2.collect()
# print(res)res = rdd3.collect()
print(res)

DAG 管理维护rdd之间依赖关系，保证代码的执行顺序，

---

DAG会根据依赖关系划分stage,每个stage都是一个独立的计算步骤，当发生宽依赖时，会单独拆分一个计算步骤（stage），进行相关数据计算，可以保证每个单独的stage可以并行执行

在发生宽依赖进行shuffle时，会独立的方法执行shuffle计算

---

拆分计算步骤的本质是为了保证数据计算的并行执行

查看spark的计算过程，通过DAG判断算子是宽依赖还是窄依赖

拆分了计算stage是宽依赖，没有拆分是窄依赖

启动spark的历史日志

start-history-server.sh

4.shuffle过程

mapreduce的shuffle作用： 将map计算后的数据传递给redue使用。
mapreduce的shuffle过程： 分区（将相同key的数据放在一个分区,采用hash），排序，合并（规约）。
将map计算的数据传递给reduce

---

spark中也有shuffle
当执行宽依赖的算子就会进行shuffle
将rdd的数据传递给下一个rdd，进行数据交换

---

无论是spark还是mr，shuffle的本质是传递交换数据。

4.1 shuffle介绍

• spark的shuffle的两个部分
  • shuffle write 写
  • shuffle read 读
  • 会进行文件的读写，影响spark的计算速度
• spark的shuffle方法类
  • 是spark封装好的处理shuffle的方法
  • hashshuffle类
    • 进行的是hash计算
    • spark1.2版本之前主要使用，之后引入了sortshuffle
    • spark2.0之后，删除了hashshuffle，从2.0版本开始使用sortshuffle类
    • 优化的hashshuffle和未优化的ashshuffle
  • sortshuffle类
    • 排序方式将相同key值数据放在一起
    • sortshuffle类使用时，有两个方法实现shuffle
      • bypass模式版本和普通模式版本
      • bypass模式版本不会排序，会进行hash操作
      • 普通模式版本会排序进行shuffle
    • 可以通过配置指定按照那种模式执行 根据task数量决定 默认 task数量小于等于200 采用bypass，task数量超过200个则使用普通模式的方法进行shuffle
    • 一个分区对应一个task，所以task数量由分区数决定

普通模式和bypass模式的主要区别在于如何将相同key值的数据放在一起

• 排序 普通模式采用的策略
• 哈希取余 bypass模式采用的策略

4.2 spark计算要尽量避免shuffle

# 优化计算，减少shuffle
from pyspark import SparkContext
sc = SparkContext()rdd = sc.parallelize([('男',20),('男',23),('女',20),('女',22)])
#求不同性别的年龄和
#reduceByKey  是宽依赖算子
rdd2 = rdd.reduceByKey(lambda x,y:x+y)# 避免shuffle，需要将宽依赖算子计算的过程换成窄依赖
boy = sc.accumulator(0)
girl = sc.accumulator(0)
def func(x):if x[0]=='男':boy.add(x[1])else :girl.add(x[1])return None
rdd3 = rdd.map(func)rdd3.collect()
print(boy.value)
print(girl.value)

5. 并行度

• 资源并行度

  • task在指定任务能够使用到的cpu核心数量

  • 多任务 多个进程或多个线程执行任务

    • 两种方式
      • 并行 多个任务同时执行
      • 并发 任务交替执行
      • 和cpu的核心数有关
        • 例如
        • cpu核心是4核 有两个线程任务 两个线程任务可以 并行执行
        • cpu核心是4核 有八个线程任务 并发执行

  • spark中cpu核心数据设置

    • –num-executors=2 设置executors数量 和服务器数量保持一致
    • –executor-cores=2 设置每个executors中的cpu核心数 每个服务器中cpu核心数一致

    spark-submit  --master yarn  --num-executors=3   --executor-cores=2
    

    最大支持的task并行数量是 num-executors* executor-cores =6

    需要按照服务器实际的cpu核心数指定 lscpu

• 数据并行度

  • 就是task数量，task由分区数决定
  • 为了保证task能充分利用cpu资源，实现并行计算，需要设置分区数应该和资源并行度一致
  • 在实际公司中就要根据公司资源并行度进行设置分区数
  • 有的场景下公司会要求数据并行度大于资源并行度

资源并行度，

按照yarn安装的服务器数量指定excutor数量 3

核心数量按照yarn中的nodemanager中的核心数指定 2
数据并行度指定

官方建议 数据并行度的task数量和资源并行度数量一致