Spark笔记(pyspark)1

简介: Spark笔记(pyspark)

Spark是什么:Spark是基于内存的迭代式计算引擎

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)

RDD的两种操作(算子)

  • 转换算子(Transformation)
  • 动作算子(Action)

flatMap(func)类似于map,但是每一个输入元素可以被映射为 0 或多个输出元素(所以 func 应该返回一个序列,而不是单一元素)

1.groupByKey和reduceByKey的区别

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

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

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

3. mapPartitions 和 foreachPartition 的区别?

mapPartitions 带有返回值;foreachPartition不带。

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

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

5、Shuffle与依赖

Shuffle就是对数据进行重组

在Spark中,有两种依赖关系:

  • 宽依赖:一个父RDD的一个分区对应一个子RDD的多个分区;
  • 窄依赖:一个父RDD的分区对应于一个子RDD的分区,或多个父RDD的分区对应于一个子RDD的分区。

窄依赖和宽依赖,主要取决于是否包含Shuffle操作。(宽依赖还有一个别名:shuffle)

窄依赖可以实现“流水线”优化。宽依赖无法实现“流水线”优化。

6、持久化

RDD的数据是过程数据,只在处理的过程中存在,一旦处理完成,就不见了:RDD之间进行相互迭代计算(Transformation的转换),当执行开启后,新的RDD生成,代表老RDD的消失。(这个特性可以最大化的利用资源,老旧RDD没用了,就从内存中清理,给后续的计算腾出空间)

在Spark中,RDD采用惰性求值的机制,每次遇到行动操作,都会从头开始执行计算。每次调用行动操作,都会触发一次从头开始的计算。这对于迭代计算而言,代价是很大的,迭代计算经常需要多次重复使用同一组数据。

可以使用persist()方法,对一个RDD标记为持久化,避免这种重复计算的开销。之所以说“标记为持久化”,是因为出现persist()语句的地方,并不会马上计算生成RDD并把它持久化,而是要等到遇到第一个行动操作触发真正计算以后,才会把计算结果进行持久化 。持久化后的RDD将会被保留在计算节点的内存中被后面的行动操作重复使用。

1698845339430.jpg

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

5.Spark中数据倾斜

数据倾斜:在任务执行期间,RDD会被分为一系列的分区,每个分区都是整个数据集的子集。当spark调度并运行任务的时候,Spark会为每一个分区中的数据创建一个任务。大部分的任务处理的数据量差不多,但是有少部分的任务处理的数据量很大,因而Spark作业会看起来运行的十分的慢,从而产生数据倾斜(进行shuffle的时候)

数据倾斜只出现在shuffle过程中,可能会触发shuffle操作的算子:distinct、groupByKey、reduceByKey、aggregateByKey、join、cogroup、repartition等

解决办法:

  • 过滤少量导致数据倾斜的key (如果发现导致倾斜的key就少数几个,而且对计算本身的影响并不大的话)
  • 提高shuffle操作的并行度(增加shuffle read task的数量,可以让原本分配给一个task的多个key分配给多个task,从而让每个task处理比原来更少的数据)
  • 两阶段聚合 局部聚合+全局聚合(将原本相同的key通过附加随机前缀的方式,变成多个不同的key,就可以让原本被一个task处理的数据分散到多个task上去做局部聚合,进而解决单个task处理数据量过多的问题。接着去除掉随机前缀,再次进行全局聚合,就可以得到最终的结果)
  • join时使用广播变量Broadcast将较小数据量广播至每个Executor(相当于reduce join转为map join)

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中的指定列(通过传入参数进行指定)

1698845353483.jpg

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注册成表,采用如下的方式:

1698845388732.jpg1698845396148.jpg

4.pyspark.sql.functions 包

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

例:

1698845406858.jpg


5.SparkSQL Shuffle 分区数目

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

1698845420181.jpg

相关文章
|
SQL 分布式计算 HIVE
pyspark笔记(RDD,DataFrame和Spark SQL)1
pyspark笔记(RDD,DataFrame和Spark SQL)
118 1
|
5月前
|
分布式计算 运维 Serverless
EMR Serverless Spark PySpark流任务体验报告
阿里云EMR Serverless Spark是一款全托管的云原生大数据计算服务,旨在简化数据处理流程,降低运维成本。测评者通过EMR Serverless Spark提交PySpark流任务,体验了从环境准备、集群创建、网络连接到任务管理的全过程。通过这次测评,可以看出阿里云EMR Serverless Spark适合有一定技术基础的企业,尤其是需要高效处理大规模数据的场景,但新用户需要投入时间和精力学习和适应。
7179 43
EMR Serverless Spark PySpark流任务体验报告
|
4月前
|
分布式计算 运维 Serverless
EMR Serverless Spark 实践教程 | 通过 EMR Serverless Spark 提交 PySpark 流任务
在大数据快速发展的时代,流式处理技术对于实时数据分析至关重要。EMR Serverless Spark提供了一个强大而可扩展的平台,它不仅简化了实时数据处理流程,还免去了服务器管理的烦恼,提升了效率。本文将指导您使用EMR Serverless Spark提交PySpark流式任务,展示其在流处理方面的易用性和可运维性。
262 7
EMR Serverless Spark 实践教程 | 通过 EMR Serverless Spark 提交 PySpark 流任务
|
5月前
|
分布式计算 运维 Serverless
通过Serverless Spark提交PySpark流任务的实践体验
EMR Serverless Spark服务是阿里云推出的一种全托管、一站式的数据计算平台,旨在简化大数据计算的工作流程,让用户更加专注于数据分析和价值提炼,而非基础设施的管理和运维。下面就跟我一起通过Serverless Spark提交PySpark流任务吧。
209 1
|
5月前
|
机器学习/深度学习 分布式计算 API
技术好文:Spark机器学习笔记一
技术好文:Spark机器学习笔记一
38 0
|
6月前
|
分布式计算 大数据 Linux
Python大数据之PySpark(三)使用Python语言开发Spark程序代码
Python大数据之PySpark(三)使用Python语言开发Spark程序代码
239 0
|
SQL 存储 分布式计算
pyspark笔记(RDD,DataFrame和Spark SQL)2
pyspark笔记(RDD,DataFrame和Spark SQL)
87 2
|
分布式计算 资源调度 Java
Spark笔记(pyspark)2
Spark笔记(pyspark)
114 0
|
SQL 机器学习/深度学习 分布式计算
spark与pyspark教程(一)
spark与pyspark教程(一)
401 0
|
分布式计算 Scala Spark
Scala写Spark笔记
Scala写Spark笔记
83 0