spark之RDD

Posted 2022-12-15 未央夜色

综述

概念与特性

RDD是spark最重要的抽象。spark统一建立在抽象的RDD之上。设计一个通用的编程抽象，使得spark可以应对各种场合的大数据情景。RDD模型将不同的组件融合到一起，选用其中的几个/所有，可以应付各种不同的场景。解决了mr的缺陷
1. 弹性分布式数据集Resilient Distributed Dataset。
2. 只读分区数据集，final修饰的
3. 一个分布式的数据集合，是spark中的核心，spark的操作都是围绕RDD展开的。
4. 真正的运算是在各个计算节点。
5. 当某个RDD操作丢失的时候，可以很快恢复。

分区

• 不同分区可能被划分到不同机器上。但是每个分区对应一个数据block
• 分区是个逻辑概念，新旧分区可能是同一块内存。（重要的优化，节约资源。）。在函数式编程，经常使用常量，但是很费内存，rdd的这种优化非常实用。防止内存的无限性扩充。
• 只是记录需要做的操作。只有当真正要执行的时候，才具体的执行。

计算

• 并行计算。计算/处理都是在各分区上，并行计算。并行，提高了效率。
• 真正的数据处理都是在各个分散的节点上。

依赖

• 子RDD从父RDD产生，父子RDD之间的关系。
  
  • 宽依赖：依赖上级所有的RDD分区。宽依赖一般非常消耗资源，结果一般要缓存下来
  • 窄依赖：依赖上级RDD的部分分区。计算的时候可能都在同一个节点上，节省资源
• stage以依赖的区别，分成不同的stage
• 每个父RDD的分区，只能被最多一个字RDD使用，子RDD可以使用任意个父RDD。

RDD的操作

创建

• 从外部数据集中读取。来源于文件系统（HDFS，HBASE），textFile方法
  这里的路径要让每个RDD的节点都能访问到

// 从外部文本中创建RDD。可以指定分片的个数
lines =  sc.textFle("路径",3)
// 一个目录下所有的文件。
lines = sc.wholeTextFile("路径")
// 从hadoop系统创建RDD
val rdd = newApiHadoopFile()
// 从hadoop数据创建RDD
val RDD = new ApiHadoopRDD()

• 从驱动程序中对一个集合进行并行化
  在测试的时候用的多，parallelize方法。可以指定分区个数

val lines = sc.parallelize(list["name","age"])

转化（Transform）

• 不进行具体操作，类似scala中的惰性求值
• 从一个RDD生成另一个RDD的过程。spark用lineage的方式表示各个RDD的依赖关系，链表的表头是textFile
• 参考fp中的概念，这里只做逻辑运算，接受一个RDD，结果产生一个RDD，没有任何副作用
• RDD常见的转化操作

map                RDD.map(fun)              将函数应用于每个元素，结果返回一个RDD
flatmap            RDD.flatmap(fun)          同map，返回一个包含所有处理结果的整体。生成的分片数不变，只是在逻辑上成一个整体
filter             RDD.filter(fun)           过滤掉不符合要求的数据 
distinct           RDD.distinct()            去重，需要shuffle代价大
union              RDD.union(RDD1)           两个RDD求并集
intersection       RDD. intersection(RDD1)   两个RDD求交集
substract          RDD.substract(RDD1)       从RDD中移除RDD1的内容
cartesian          RDD.cartesian(RDD1)       生成RDD与RDD1的笛卡尔积
pipe               RDD.pipe("shell命令")      利用linux中的shell语言，对数据操作。
zip                RDD.zip(RDD1)             将RDD和RDD1组成一个kv格式的新RDD
sample             RDD.sample()              随机采样，返回RDD
-----------------------------------------------------------------------------
groupBy            RDD.groubBy(key)          根据key的取值，分组
reduceByKey        RDD.reduceByKey()         根据key的取值聚合
sortByKey          RDD.sortByKey()           根据key的值排序

行动（Action）

• 真正的开始处理和操作，强制执行所有的RDD
• RDD常见的行动操作

collect()           RDD.collect()             返回RDD中的所有元素。只能计算小规模的RDD，否则要shuffle代价大
count()             RDD.count()               统计RDD中元素的个数
countByVale()       RDD.countByValue()        每个元素在RDD中出现的次数。
take()              RDD.take(n)               返回RDD中的n个元素
top()               RDD.top(N)                返回RDD中的前N个元素
takeOrdered()       RDD.takeOrdered(n)        按照要求的顺序返回前n个元素
takeSample()        RDD.takeSample(n)         从RDD中任意返回n个元素
reduce()            RDD.reduce(fun)           并行整合RDD中所有的元素
fold()              RDD.fold(num)(fun)        提供初始值，的reduce
aggregate()         RDD.aggregate()            ？？？？？？？？？
foreach()           RDD.foreach(fun)          对RDD中的 每个元素使用给定的函数

RDD常见操作

过滤

val textRDD = sc.textFile(path)
val resultRDD1 = textRDD.filter(_.split(" ")(0).equals("1"))
val resultRDD2 = resultRDD1.map(s =>  | val columns = s.split(",") | val pm = columns(1)  | val host = columns(2).replaceAll("a","b") pm+""+host )

去重

val RDD = sc.textFile("path")
val R1 = RDD.map(s =>  | val one = s.split(",") | val two = columns(0) | val three = columms(1) | (p2,p1) | )
val R2 = R1.distinct()

共享变量

广播变量

副本拷贝的方式获得

累加器

只能进行add操作

持久化

• 提高了数据的可重用性
• 把RDD中的结果持久化到内存中。当后续的操作需要用到某些RDD运算结果的时候，持久化到内存可以提高效率。主要有cahce方法和persist方法。
• cache方法只有一个默认的缓存级别（MEMORY_ONLY），persist可以有许多级别。cache实际上是调用了persist方法
• 当要缓存的内容太多，用LRU算法淘汰。
• 保存

RDD.saveAsTextFile("路径")

• 持久化级别：

级别	描述
MEMORY_ONLY	只放在内存，超过的部分丢弃，不存储（默认的级别）
MEMORY_AND_DISK	放在内存，超过的放在磁盘上
MEMORY_ONLY_SER	RDD作为序列化的对象存储
MEMORY_AND_DISK_SER	超过的部分放在disk上
DISK_ONLY	只放在disk上

工作流程

• RDD把操作记录程DAG图，记录各个DAG中的转换关系
• 无论进行了多少次转换，只有真正遇到action的时候才真正计算