Pyspark学习笔记（四）---弹性分布式数据集 RDD [Resilient Distribute Data]（上）

Pyspark学习笔记（四）---弹性分布式数据集 RDD [Resilient Distribute Data] （上）

• 1.RDD简述
• 2.加载数据到RDD
• • A 从文件中读取数据
  • • Ⅰ·从文本文件创建RDD
    • Ⅱ·从对象文件创建RDD
  • B 从数据源创建RDD
  • C.通过编程创建RDD
• 3.RDD操作
• 4.RDD持久化与重用
• 5.RDD谱系
• 6.窄依赖（窄操作）- 宽依赖（宽操作）：
• 7.RDD容错性
• 8.RDD类型
• 9.基本的RDD操作

1.RDD简述

RDD是Spark编程中最基本的数据对象， 无论是最初加载的数据集，还是任何中间结果的数据集，或是最终的结果数据集，都是RDD。 在Pyspark中，RDD是由分布在各节点上的python对象组成，如列表，元组，字典等。 RDD主要是存储在内存中（亦可持久化到硬盘上），这就是相对于Hadoop的MapReduce的优点，节省了重新读取硬盘数据的时间。

• 弹性：RDD是有弹性的，意思就是说如果Spark中一个执行任务的节点丢失了，数据集依然可以被重建出来；
• 分布式：RDD是分布式的，RDD中的数据被分到至少一个分区中，在集群上跨工作节点分布式地作为对象集合保存在内存中；
• 数据集：RDD是由记录组成的数据集。

RDD的另一个关键特性是不可变，也即是在实例化出来导入数据后，就无法更新了。 每次对已有RDD进行转化操作（transformation）都会生成新的RDD；

2.加载数据到RDD

要开始一个Spark程序，需要从外部源的数据初始化出至少一个RDD。 然后才是经过一系列转化操作、行动操作，得到中间的RDD和结果RDD。 初始RDD的创建方法：

• A 从文件中读取数据；
• B 从SQL或者NoSQL等数据源读取
• C 通过编程加载数据
• D 从流数据中读取数据。

#创建一个SparkSession对象，方便下面使用
from pyspark.sql import SparkSession
spark = SparkSession\
                    .builder\
                    .appName('exam1')\
                    .enableHiveSupport()\
                    .getOrCreate()
 sc = spark.sparkContext

A 从文件中读取数据

Ⅰ·从文本文件创建RDD

sc.textFile(name, minPartitions=None, use_unicode=True)
#示例：
#①读取整个目录下的内容
Example=sc.textFile(“hdfs://exam_dir/running_logs/”)

#②读取目录下的单个文件
Example=sc.textFile(“hdfs://exam_dir/running_logs/log_001.txt”)

#③使用通配符读取文件
Example=sc.textFile(“hdfs://exam_dir/running_logs/*_001.txt”)
#####################################################################
sc.wholeTextFiles(path, minPartitions=None, use_unicode=True)
#读取包含多个文件的整个目录，每个文件会作为一条记录（键-值对）；
#其中文件名是记录的键，而文件的全部内容是记录的值。
#使用textFile()读取目录下的所有文件时，每个文件的每一行成为了一条单独的记录，
#而该行属于哪个文件是不记录的。

Ⅱ·从对象文件创建RDD

对象文件指序列化后的数据结构，有几个方法可以读取相应的对象文件： hadoopFile(), sequenceFile(), pickleFile()

B 从数据源创建RDD

一般是使用SparkSession中的函数，SparkSession对象提供了read method，返回一个DataFrameReader对象。官网链接如下 http://spark.apache.org/docs/latest/api/python/reference/pyspark.sql.html#pyspark.sql.SparkSession.read 用该对象将数据读取到DataFrame中，DataFrame是一种特殊的RDD，老版本中称为SchemaRDD。 比如说，spark现在是一个已经被创建的SparkSession对象，然后调用read方法，spark.read就是一个DataFrameReader对象，然后就调用该对象(DataFrameReader)的一系列方法，来读取各种数据,参考如下链接：http://spark.apache.org/docs/2.3.0/api/java/org/apache/spark/sql/DataFrameReader.html

官方github中给了很多例子： https://github.com/apache/spark/blob/master/examples/src/main/python/sql/datasource.py

C.通过编程创建RDD

sc.parallelize(c, numSlices=None)

parallelize()方法要求列表已经创建好，并作为c参数传入。参数numSlices指定了所需创建的分区数量。 http://spark.apache.org/docs/latest/api/python/_modules/pyspark/context.html#SparkContext.parallelize

3.RDD操作

转化操作：操作RDD并返回一个 新RDD 的函数； 行动操作：操作RDD并返回 一个值 或者 进行输出 的函数。

粗粒度转化操作：把函数作用于数据的每一个元素（无差别覆盖），比如map，filter 细粒度转化操作：可以针对单条记录或单元格进行操作。

惰性求值 在处理Spark程序时，Spark使用惰性求值（lazy evaluation），也叫做惰性执行（lazy execution)。惰性执行指的 是在调用行动操作时（也就是需要进行输出时）再处理数据。这是因为每个语句仅仅解析了语法和引用对象， 在请求了行动操作之后，Spark会创建出DAG图以及逻辑执行计划和物理执行计划，接下来驱动器进程就跨执行器协调并管理计划的执行。

4.RDD持久化与重用

RDD主要创建和存在于执行器的内存中。默认情况下，RDD是易逝对象，仅在需要的时候存在。 在它们被转化为新的RDD，并不被其他操作所依赖后，这些RDD就会被删除。 若一RDD在多个行动操作中用到，就每次都会重新计算，则可调用cache()或persist( )方法缓存或持久化RDD。

5.RDD谱系

Spark维护每个RDD的谱系，也就是获取这个RDD所需要的一系列转化操作的序列。 默认情况下，每个RDD都会重新计算整个谱系，除非调用了RDD持久化。

6.窄依赖（窄操作）- 宽依赖（宽操作）：

• 窄操作：
• ①多个操作可以合并为一个阶段，比如同时对一个数据集进行的map操作或者filter操作可以在数据集的各元 素的一轮遍历中处理；
• ②子RDD只依赖于一个父RDD
• ③不需要进行节点间的数据混洗
• 宽操作：
• ①通常需要数据混洗
• ②RDD有多个依赖，比如在join或者union的时候

7.RDD容错性

因为每个RDD的谱系都被记录，所以一个节点崩溃时，任何RDD都可以将其全部分区重建为原始状态。（当 然，如果存在一些非确定性函数，比如random，因为其随机性，所以可能影响到RDD的重建。）

8.RDD类型

除了包含通用属性和函数的基本类型BaseRDD外，RDD还有以下附加类型： http://spark.apache.org/docs/2.3.0/api/java/org/apache/spark/rdd/RDD.html

常见的RDD类型： PairRDD： 由键值对组成的RDD，比如前面提到的用wholeTextFiles()方法读取的内容就是以键值对的形式存在 DoubleRDD: 由双精度浮点数组成的RDD。 DataFrame：以前的版本被称为SchemaRDD，按一组有固定名字和类型的列来组织的分布式数据集。DataFrame等价于sparkSQL中的关系型表! 所以我们在使用sparkSQL的时候常常要创建这个DataFrame，在sparkSQL部分会提及。 HadoopRDD：提供读取存储在HDFS上的数据的RDD。

9.基本的RDD操作

Pyspark学习笔记（四）—弹性分布式数据集 RDD 【Resilient Distribute Data】（下）