Spark笔记（pyspark）

Spark笔记

• • • 1、基本概念
    • 2、架构设计
    • 3、Spark运行流程
    • 4、弹性分布数据集(RDD)
    • • 1.groupByKey和reduceByKey的区别
      • 2. 哪两个Action算子的结果不经过Driver, 直接输出?
      • 3. mapPartitions 和 foreachPartition 的区别?
    • 5、Shuffle与依赖
    • 6、持久化
    • • 1. Cache和Checkpoint区别
      • 2. Cache 和 CheckPoint的性能对比?
    • 7、Spark On Yarn两种模式总结
    • 8、Spark内核调度
    • • 1.DAG之Job和Action
      • 2.Spark是怎么做内存计算的？DAG的作用？Stage阶段划分的作用？
      • 3. Spark为什么比MapReduce快
      • 4.Saprk并行度
    • 9、DataFrame
    • • 1.DataFrame的组成
      • 2.DataFrame之DSL
      • 3.DataFrame之SQL
      • 4.pyspark.sql.functions 包
      • 5.SparkSQL Shuffle 分区数目
      • 6.SparkSQL 数据清洗API
      • 7.DataFrame数据写出
    • 10、SparkSQL
    • • 1.定义UDF函数
      • 2.使用窗口函数

1、基本概念

RDD：是Resillient Distributed Dataset（弹性分布式数据集）的简称，是分布式内存的一个抽象概念，提供了一种高度受限的共享内存模型

DAG：是Directed Acyclic Graph（有向无环图）的简称，反映RDD之间的依赖关系

Executor：是运行在工作节点（WorkerNode）的一个进程，负责运行Task

应用（Application）：用户编写的Spark应用程序

任务（ Task ）：运行在Executor上的工作单元

作业（ Job ）：一个作业包含多个RDD及作用于相应RDD上的各种操作

阶段（ Stage ）：是作业的基本调度单位，一个作业会分为多组任务，每组任务被称为阶段，或者也被称为任务集合，代表了一组关联的、相互之间没有Shuffle依赖关系的任务组成的任务集

2、架构设计

Spark运行架构包括集群资源管理器（Cluster Manager）、运行作业任务的工作节点（Worker Node）、每个应用的任务控制节点 （Driver）和每个工作节点上负责具体任务的执行进程（Executor）。资源管理器可以自带或Mesos或YARN 。
在Spark中，一个应用（Application）由一个任务控制节点（Driver）和若干个作业（Job）构成，一个作业由多个阶段（Stage）构成，一个阶段由多个任务（Task）组成。当执行一个应用时，任务控制节点会向集群管理器（Cluster Manager）申请资源，启动Executor，并向Executor发送应用程序代码和文件，然后在Executor上执行任务，运行结束后，执行结果会返回给任务控制节点，或者写到HDFS或者其他数据库中。

3、Spark运行流程

Spark运行的4个步骤：

• 当一个Spark应用被提交时，Driver创建一个SparkContext，由SparkContext负责和资源管理器（Cluster Manager）的通信以及进行资源的申请、任务的分配和监控等。SparkContext会向资源管理器注册并申请运行Executor的资源 ；
• 资源管理器为Executor分配资源，并启动Executor进程，Executor发送心跳到资源管理器上；
• SparkContext根据RDD的依赖关系构建DAG图，DAG图提交给DAG调度（DAGScheduler）进行解析，将DAG图分解成多个“阶段”，并且计算出各个阶段之间的依赖关系，然后把一个个“任务集”提交给底层的任务调度器（TaskScheduler）进行处理；Executor向SparkContext申请任务，任务调度器将任务分发给Executor运行，同时，SparkContext将应用程序代码发放给Executor；
• 任务在Executor上运行，把执行结果反馈给任务调度器，然后反馈给DAG调度器，运行完毕后写入数据并释放所有资源 。

4、弹性分布数据集(RDD)

1.groupByKey和reduceByKey的区别

reduceByKey自带聚合逻辑, groupByKey不带；如果做数据聚合reduceByKey的效率更好, 因为可以先聚合后shuffle再最终聚合, 传输的IO小。

2. 哪两个Action算子的结果不经过Driver, 直接输出?

foreach 和 saveAsTextFile 直接由Executor执行后输出，不会将结果发送到Driver上去。

3. mapPartitions 和 foreachPartition 的区别?

mapPartitions 带有返回值；foreachPartition不带。

转换得到的RDD是惰性求值的。也就是说，整个转换过程只是记录了转换的轨迹，并不会发生真正的计算，只有遇到行动操作时，才会发生真正的计算，开始从血缘关系源头开始，进行物理的转换操作。行动操作是真正触发计算的地方。Spark程序执行到行动操作时，才会执行真正的计算，从文件中加载数据，完成一次又一次转换操作，最终，完成行动操作得到结果。

所以遇到错误时，不一定是行动的原因，可能是之前的某个地方的错误，要看错误原因是什么

5、Shuffle与依赖

在Spark中，有两种依赖关系：

• 宽依赖：一个父RDD的一个分区对应一个子RDD的多个分区；
• 窄依赖：一个父RDD的分区对应于一个子RDD的分区，或多个父RDD的分区对应于一个子RDD的分区。

窄依赖和宽依赖，主要取决于是否包含Shuffle操作。(宽依赖还有一个别名：shuffle)
窄依赖可以实现“流水线”优化。宽依赖无法实现“流水线”优化。

6、持久化

RDD的数据是过程数据，只在处理的过程中存在，一旦处理完成，就不见了：RDD之间进行相互迭代计算（Transformation的转换），当执行开启后，新的RDD生成，代表老RDD的消失。（这个特性可以最大化的利用资源，老旧RDD没用了，就从内存中清理，给后续的计算腾出空间）
在Spark中，RDD采用惰性求值的机制，每次遇到行动操作，都会从头开始执行计算。每次调用行动操作，都会触发一次从头开始的计算。这对于迭代计算而言，代价是很大的，迭代计算经常需要多次重复使用同一组数据。
可以使用persist()方法，对一个RDD标记为持久化，避免这种重复计算的开销。之所以说“标记为持久化”，是因为出现persist()语句的地方，并不会马上计算生成RDD并把它持久化，而是要等到遇到第一个行动操作触发真正计算以后，才会把计算结果进行持久化 。持久化后的RDD将会被保留在计算节点的内存中被后面的行动操作重复使用。

1. Cache和Checkpoint区别

Cache是轻量化保存RDD数据, 可存储在内存和硬盘, 是分散存储, 设计上数据是不安全的(保留RDD 血缘关系)
CheckPoint是重量级保存RDD数据, 是集中存储, 只能存储在硬盘(HDFS)上, 设计上是安全的(不保留 RDD血缘关系)

2. Cache 和 CheckPoint的性能对比?

Cache性能更好, 因为是分散存储, 各个Executor并行执行, 效率高, 可以保存到内存中(占内存),更快
CheckPoint比较慢, 因为是集中存储, 涉及到网络IO, 但是存储到HDFS上更加安全(多副本)

7、Spark On Yarn两种模式总结

Client模式和Cluster模式最最本质的区别是:Driver程序运行在哪里。

• Client模式:学习测试时使用，生产不推荐(要用也可以,性能略低,稳定性略低)
  1.Driver运行在Client上,和集群的通信成本高
  2.Driver输出结果会在客户端显示
• Cluster模式:生产环境中使用该模式
  1.Driver程序在YARN集群中，和集群的通信成本低
  2.Driver输出结果不能在客户端显示
  3.该模式下Driver运行ApplicattionMaster这个节点上,由Yarn管理，如果出现问题，yarn会重启 ApplicattionMaster(Driver)

8、Spark内核调度

1.DAG之Job和Action

1个Action会产生1个DAG，如果在代码中有3个Action就产生3个DAG；一个Action产生的一个DAG，会在程序运行中产生一个JOB，所以：1个ACTION = 1个DAG= 1个JOB。
如果一个代码中，写了3个Action，那么这个代码运行起来产生3个JOB，每个JOB有自己的DAG；一个代码运行起来，在Saprk中称之为：Application。
层级关系：1个Application中，可以有多个JOB，每一个JOB内含一个DAG，同时每一个JOB都是由一个Action产生的。

2.Spark是怎么做内存计算的？DAG的作用？Stage阶段划分的作用？

1. Spark会产生DAG图
2. DAG图会基于分区和宽窄依赖关系划分阶段
3. 一个阶段的内部都是窄依赖，窄依赖内，如果形成前后1：1的分区对应关系，就可以产生许多内存迭代计算的管道。
4. 这些内存迭代计算的管道，就是一个个具体的执行Task
5. 一个Task是一个具体的线程，任务跑在一个线程内，就是走内存计算了。

3. Spark为什么比MapReduce快

1. Spark的算子丰富，MapReduce算子匮乏（Map和Reduce），MapReduce这个编程模型，很难在一套MR中处理复杂的任务，很多复杂的任务，是需要写多个MapReduce进行串联，多个MR串联通过磁盘交互数据
2. Saprk可以执行内存迭代，算子之间形成DAG，基于依赖划分阶段后，在阶段内形成内存迭代管道，但是MApReduce的Map和Reduce之间的交互依旧是通过硬盘来交互的。

4.Saprk并行度

全局并行度配置的参数：spark.default.parallelism

9、DataFrame

1.DataFrame的组成

在结构层面:

• StructType对象描述整个DataFrame的表结构
• StructField对象描述一个列的信息

在数据层面

• Row对象记录一行数据
• Column对象记录一列数据并包含列的信息

2.DataFrame之DSL

"""
1. agg: 它是GroupedData对象的API, 作用是 在里面可以写多个聚合
2. alias: 它是Column对象的API, 可以针对一个列 进行改名
3. withColumnRenamed: 它是DataFrame的API, 可以对DF中的列进行改名, 一次改一个列, 改多个列 可以链式调用
4. orderBy: DataFrame的API, 进行排序, 参数1是被排序的列, 参数2是 升序(True) 或 降序 False
5. first: DataFrame的API, 取出DF的第一行数据, 返回值结果是Row对象.
# Row对象 就是一个数组, 你可以通过row['列名'] 来取出当前行中, 某一列的具体数值. 返回值不再是DF 或者GroupedData 或者Column而是具体的值(字符串, 数字等)
"""

1.show方法
功能:展示DataFrame中的数据, 默认展示20条

df.show(参数1, 参数2)
- 参数1: 默认是20, 控制展示多少条
- 参数2: 是否阶段列, 默认只输出20个字符的长度, 过长不显示, 要显示的话请填入truncate=True

2.printSchema方法
功能:打印输出df的schema信息

df.printSchema()

3.select
功能:选择DataFrame中的指定列(通过传入参数进行指定)

4. filter和where
功能:过滤DataFrame内的数据，返回一个过滤后的DataFrame
5.groupBy 分组
功能:按照指定的列进行数据的分组， 返回值是GroupedData对象

df.groupBy()

传入参数和select一样，支持多种形式。GroupedData对象是一个特殊的DataFrame数据集，GroupedData对象也有很多API，比如count、min、max、avg、sum等等

3.DataFrame之SQL

如果想使用SQL风格的语法，需要将DataFrame注册成表,采用如下的方式:

4.pyspark.sql.functions 包

里的功能函数, 返回值多数都是Column对象.
例：

5.SparkSQL Shuffle 分区数目

在SparkSQL中当Job中产生产生Shuffle时，默认的分区数（spark.sql.shuffle.partitions）为200，在实际项目中要合理的设置。可以设置在：

6.SparkSQL 数据清洗API

1.去重方法 dropDuplicates
功能：对DF的数据进行去重，如果重复数据有多条，取第一条

2.删除有缺失值的行方法 dropna
功能：如果数据中包含null，通过dropna来进行判断，符合条件就删除这一行数据

3.填充缺失值数据 fillna
功能：根据参数的规则，来进行null的替换

7.DataFrame数据写出

spark.read.format()和df.write.format() 是DataFrame读取和写出的统一化标准API
SparkSQL 统一API写出DataFrame数据

DataFrame可以从RDD转换、Pandas DF转换、读取文件、读取 JDBC等方法构建

10、SparkSQL

1.定义UDF函数

方式1语法:
udf对象 = sparksession.udf.register(参数1，参数2，参数3)

• 参数1:UDF名称，可用于SQL风格
• 参数2:被注册成UDF的方法名
• 参数3:声明UDF的返回值类型

udf对象: 返回值对象，是一个UDF对象，可用于DSL风格
方式2语法:
udf对象 = F.udf(参数1， 参数2)

• 参数1:被注册成UDF的方法名
• 参数2:声明UDF的返回值类型

udf对象: 返回值对象，是一个UDF对象，可用于DSL风格
其中F是:from pyspark.sql import functions as F 其中，被注册成UDF的方法名是指具体的计算方法，如: def add(x, y): x + y
add就是将要被注册成UDF的方法名

2.使用窗口函数

开窗函数
开窗函数的引入是为了既显示聚集前的数据，又显示聚集后的数据。即在每一行的最后一列添加聚合函数的结果。
开窗用于为行定义一个窗口(这里的窗口是指运算将要操作的行的集合)，它对一组值进行操作,不需要使用GROUP BY子句对数据进行分组，能够在同一行中同时返回基础行的列和聚合列。
聚合函数和开窗函数
聚合函数是将多行变成一行， count,avg…
开窗函数是将一行变成多行;
聚合函数如果要显示其他的列必须将列加入到group by中
开窗函数可以不使用group by,直接将所有信息显示出来
开窗函数分类
1.聚合开窗函数
聚合函数(列) OVER(选项)，这里的选项可以是PARTITION BY子句、但不可以是ORDER BY子句。
2.排序开窗函数
排序函数(列) OVER(选项)，这里的选项可以是ORDER BY子句，也可以是OVER(PARTITION BY子句ORDER BY子句)，但不可以是PARTITION BY子句。
3.分区类型NTILE的窗口函数