spark基本原理&UI界面解读

这里是引用

1 八股文

1.1 基本原理

driver节点是整个应用程序的指挥所
指挥官是sparkcontext
环境：构建一个集群
应用程序提交
确定主节点，确定指挥所driver，确定指挥官sparkcontext
sparkcontext会向资源管理器申请资源
会将作业分为不同阶段
将不同任务分到不同节点执行
整个过程还会进行监控
资源管理器收到sparkcontext的资源请求
会向executor分配资源
启动executor进程，才会启动线程
executor进程是驻留在不同的work node中
会有成百上千个进程和work node
sparkcontext对象要根据 rdd依赖关系构建一个DAG图
代码就是针对RDD一次次的操作
这些操作会被转换成一个有向无环图 dag
DAG会被提交到dag scheduler解析
DAG图会被切为很多个阶段 stage
每个stage又分为若干个任务
每一个阶段stage是任务的集合
把这个阶段stage提交给task scheduler
task scheduler负责分发任务
worker node上的executor会向task scheduler主动申请
task scheduler会返回任务给worker node上的executor去派生线程去执行
计算给节点的分发原则：
计算向数据靠拢。数据在哪个节点上面，task scheduler优先分配，完成本地化的处理。
executor运行的结果会再次反馈给task scheduler
再向上传给 dag scheduler
spark context做最后的处理。返回给用户看或者写入HDFS
sparkcontext：代表了整个应用程序连接集群的通道。链接应用和集群

1.2 核心原理

1.2.1 关键概念

driver：该进程调用spark程序的main方法，并且启动sparkcontext

cluster Manager：该进程负责和外部集群工具打交道，申请或释放集群资源

Woker：该进程是一个守护进程，负责启动和管理executor

executor:该进程是一个JVM虚拟机。负责运行spark task

1.2.2 运行步骤

启动driver，创建sparkcontext
client提交程序给driver，driver向cluster manager申请集群资源
资源申请完毕，在worker中启动executor
driver将程序转化为tasks，分发给executor执行

1.2.3 spark运行模型

1.2.3.1 单机

使用线程模拟并行来运行程序

一般用于测试开发
不能启动spark的master、worker守护进程
不能启动Hadoop的各项服务

sparksubmit进程是客户提交任务的client进程，又是spark的driver程序、还充当着spark执行task的executor的角色

1.2.3.2 集群

使用集群管理器来和不同类型的集群交互，将任务运行在集群中。

spark standalone
hadoop yarn
Apache mesos

kubernetes

2 Spark UI深入解读

在日常的开发工作中，我们总会遇到spark应用运行失败，或者执行效率不达预期的情况。对于这样的问题，想找到根本原因(root cause)，就可以通过spark ui来获取最直接、最直观的线索，在全量的审查spark应用的同时，迅速定位问题所在。

如果我们把失败的、或是执行低效的spark应用看做是“病人”的话，那么spark ui中关于应用的众多衡量指标(metrics)，就是这个病人的“体检报告”。结合多样的metrics，身为“大夫”的开发者可以结合经验来迅速定位“病灶”。

2.1 spark UI一级入口

打开spark UI，映入眼帘的是默认的Jobs页面。JObs页面记录着应用中涉及的actions动作，以及与数据读取、移动有关的动作。其中每一个action都对应着一个job，而每一个job都对应着一个作业。我们一会再去对jobs页面做展开，现在先把目光集中在spark ui最上面的导航条，这里面罗列这spark ui的所有一级入口

入口页	包含内容。作用
jobs	actions，以及数据的读取与移动等操作。作业详情预览
stages	DAG中每个stage的入口。作业详情预览
storage	分布式数据集缓存(cache)详情页。审查cache在内存与磁盘中的分布情况
storage	分布式数据集缓存(cache)详情页。审查cache在内存与磁盘中的分布情况
environment	配置项，环境变量详情。审查spark配置项是否符合预期
executors	分布式运行环境与计算负载详情页。深入审查执行计划中的每一个环节

JOB：作业详情的预览

什么情况下spark会产生一个job？对应的action算子会产生，对应的数据的读取和移动操作。
stage。一个Job可以拆分为多个stage。job拆分stage的详细情况。划分stage的依据是宽窄依赖
storage。在cache数据的时候，会有应用。
environment。spark任务的配置项、环境变量。
executor。分布式运行环境计算负载的详情页，作用是审查executor之间是否存在数据倾斜。比如，看input的数据的差异用于评估负载情况。

SQL。spark SQL执行计划的详情页。

executor、environment、storage不存在二级入口，但是SQL、storage、JOBs有二级入口

2.1.1 executors

2.1.1.1 summary

RDD blocks：原始分区数据集的分区数量
storage memory：用于cache时所占的内存的占用情况。
disk used。计算过程中消耗的磁盘空间。
cores。用于计算的CPU的核数。
acitive task：活跃的task数量
failed task：失败的task数量
complete task：完成的task数量
task time(GC time)。任务的执行时候，以及任务的GC时间
input。输入数据量的大小
shuffle read 大小
shuffle write 大小

blocklisted。黑名单

2.1.1.2 executors

executors tab的主要内容如下，主要包含“summary”和“executors”两部分。这两部分记录的度量指标是一致的，其中“executors”以更新粒度记录者每一个executor的详情，而第一部分“summary”是下面所有executors度量指标的简单加和。

sparkUI都提供了哪些metrics，来量化每一个executor的工作负载（workload）。

metric	含义
RDD	原始数据集的分区数量
storage memory	用于cache的内存占用
disk used	计算过程中消耗的磁盘空间
cores	用于计算的CPU核数
active/failed/complete	total tasks
task time(GC time)	任务执行时间(括号内为任务GC时间)
input	输入数据量大小
shuffle head/write	shuffle读写过程中消耗的数据量
logs/thread dump	日志与core dump

不难发现，executors页面清清楚楚的记录着每一个executor消耗的数据量，以及他们对CPU、内存与磁盘等硬件资源的消耗。基于这些信息，我们可以轻松判断应用中是否存在数据倾斜的隐患。

Thread Dump。Java中的诊断工具，每个JVM都可以显示所有线程在某一个点的状态，用作Java定位问题的诊断功能。

runnable。当前可以运行的线程
timed-waiting。线程主动等待的意思。

waiting。等待的线程。

summary是executors所有指标聚合的情况。

基于这些信息，盘点不同executor之间是否存在负载不均衡的情况、数据倾斜的隐患。

2.1.2 environment

各种各样环境变量与配置信息。

metric	含义
runtime information	Java Scala版本号信息
spark properties	所有spark配置项的设置细节，重点
Hadoop properties	hadoop配置项细节
system properties	应用提交方法(spark-shell/spark-submit)
classpath entries	classpath路径设置信息

spark properties是重点。其中记录着所有运行时生效的spark配置项设置。通过spark properties，我们可以确认运行时的设置，与我们预期的设置是否一致，从而排除因配置项设置错误而导致的稳定性或是性能问题。

2.1.2.1 runtime information

2.1.2.2 spark properties

spark任务的各种配置项、判断参数是否合理

2.1.2.3 resource properties

不重要

2.1.2.4 Hadoop properties

Hadoop的各种配置项

2.1.2.5 system properties

系统配置项，可以看启动命令。sum.java.command

2.1.2.6 classpath properties

配置、jar包的路径

2.1.3 storage

记录了每一个缓存，rdd cache、dataframe cache。包括缓存级别、已缓存的分区数、缓存比例、内存大小与磁盘大小。

spark支持不同的缓存级别，他是存储介质(内存、磁盘)、存储形式(对象、序列化字节)与副本数量的排列组合。对于data frame来说，默认的级别是单副本的disk memory deserialized，也就是存储介质为内存加磁盘，粗出形式为对象的单一副本存储方式。

metric	含义
storage level	存储级别
cached partitions	已缓存的分区数
fraction caches	缓存比例
size in memory	内存大小
size on disk	磁盘大小

cached partitions 和fraction caches分别记录着数据集成功缓存的分区数量，以及这些缓存的分区占所有分区的比例。当fraction cached小于100%的时候。说明分布式数据集并没有完全缓存到内存(或是磁盘)。对于这种情况，我们要警惕缓存换入换出可能带来的性能隐患。
基于storage页面提供的详细信息，我们可以有的放失的设置于内存有关的配置项，如spark.executor.memory、spark.executor.fraction、spark.executor.storageFraction、从而有针对性的对storage memory进行调整。
cache partitions 已缓存的分区数

fraction cached。缓存的比例，代表缓存的分区占所有分区的比例，当小于100%的时候，代表分布式的数据没有完全划分到内存或者磁盘里面。

缓存换入换出，有可能带来性能的问题。

size in memory。内存缓存的大小

storage memory不足的情况下，会把他size到磁盘里面。

size in disk。磁盘缓存的大小

2.1.4 SQL

以actions为单位，记录着每个action对应的sparksql执行计划。我们需要点击“description”列中的超链接，才能进入到二级页面，去了解每个执行计划的详细信息。
Jobs：同理，低于jobs来说，spark ui也是以actions为粒度，记录着每个action对应作业的执行情况。我们要了解作业详情页，也必须通过“description”页面提供的二级入口链接。
一个action对应一个query，一个query会有多个job id。

以actions为单位，记录着每个action对应的spark sql执行计划。我们需要点击“description”列中的超链接，才能进入到二级页面，去了解每个执行计划的详细信息。

2.1.4 JOBs

description。描述
submitted。提交时间
duration。执行时间

stage。成功

2.1.4 stage

我们知道，每一个作业，都包含多个阶段，就是我们常说的stages。在stages页面，spark ui罗列了应用中涉及的所有stage，这些stages分属于不同的作业。要想查看哪些stages隶属于哪个job，还需要从jobs的descritions二级入口进入查看。

stage页面，更多的是一种预览，要想查看每一个stage的详情，同样需要从“description”进入详情页。

总结：

一级入口	重点内容
executors	不同executors之间，是否存在负载倾斜
environment	不同executors之间，是否存在负载倾斜
storage	分布式数据集的缓存级别，内存，磁盘缓存比例
SQL	初步了解不同执行计划的执行时间，确实是否符合预期
jobs	初步感知不同jobs的执行时间，确实是否符合预期
stage	初步感知不同stage的执行时间，确实是否符合预期

记录了以action为粒度，记录了每个action作业的情况。
executor可以看到不同executor负载情况、执行情况，判断数据倾斜

environment可以看到spark任务的配置情况，判断配置是否合理。参数的配置。

配置优先级：code>conf>默认
SQL。可以了解不同任务执行时间是否符合预期。
job。可以看到job的执行情况，是否符合预期
storage。可以看到storage的执行情况，是否符合预期

stage。可以看到stage的执行情况，是否符合预期

2.2 spark UI二级入口

所为二级入口，指的是通过一次超链接跳转才能访问到的页面。对于SQL、jobs和stages这三类入口来说，二级入口往往已经提供了足够的信息，基本覆盖了“体检报告”的全部内容。因此，尽管spark UI也提供了少量的三级入口(需要凉调才能到达的页面)，但是这些隐藏在“犄角旮旯”的三级入口，往往不需要开发者去特别关注。

接下来我们就沿着sql->job->stages的顺序，一次的去访问他们的耳机入口，从而针对全局dag，作业以及执行阶段，获得更加深入的探索和洞察。

2.2.1 sql详情页

在 SQLtab一级入口，我们看到有1个条目

点击图中的“description”，即可进入到该作业的执行计划页面，如下图所示。

2.2.1.1 exchange

可以看到，对应每一个exchange，spark ui都提供了丰富的metrics来刻画shuffle的计算过程。从shuffle write到shuffle read，从数据量到处理时间，应有尽有。

metrics	含义
shuffle records written	shuffle write阶段写入的数据条目数量
shuffle write time total	shuffle write阶段花费的写入时间
records read	shuffle read阶段读取的数据条目数量
local bytes read total	shuffle read阶段从本地节点读取的数据总量
fetch wait time total	shuffle read阶段花费在网络传输上的时间
remote bytes read total	shuffle read阶段跨网络、从远端节点读取的数据总量
date size total	原始数据在内存中展开之后的总大小
remote bytes read to disk	shuffle read阶段因数据块过大而直接落盘的情况
shuffle bytes written total	shuffle中间文件总大小

结合这份shuffle的体检报告，我们就能一量化的方式，去掌握shuffle过程的计算细节，从而为调优提供更多的洞察和思路。

2.2.1.2 sort

接下来，我们再来说说sort。相比exchange，sort的度量指标没有那么多，不过，他们足以让我们一窥sort再运行时，对内存的消耗，如下图所示。

metrics	含义
sort time total	排序消耗的总时间
peak memory total	内存的消耗峰值(集群范围内)
spill size total	排序过程中移除到磁盘的数据总量

可以看到“peak memory total ”和“spill size total”这两个数值，足以指导我们更有针对性的去设置spark.executor.memory、spark.memory.fraction、spark.memory.storageFraction，从而使得execution memory区域得到充分的保障。

2.2.1.2 aggragate

与sort类似，衡量aggregate的度量指标，主要记录的也是操作的内存消耗。
-

可以看到对于aggregate操作，spark ui也记录着磁盘移除与峰值消耗，即spill size和peak memory total。这两个数值也为内存的调整提供了依据。

2.2.2 jobs详情页

接下里，我们再来说说jobs详情页。jobs详情页非常的简单、直观，他罗列这隶属于当前job的所有stages。要想访问每一个stage的执行细节，我们还需要通过description的超链接做跳转。

2.2.3 stages详情页

实际上，要访问stage详情，我们还有另外一种选择，那就是直接从stages以及入口进入，然后完成跳转。因此stage详情页也归类为二级入口。
接下来我们以id为8的stage位例，去看一看详情页都记录着哪些关键信息。
在所有的二级入口中，stage详情页的信息流可以说是巨大的。点击stage详情页，可以看到他主要包含3大信息类。分别是stage dag、event timeline与task metrics.

其中task metics又分为summary和entry detail两部分，提供不同粒度的信息汇总。而task metrics中记录的指标类别，还可以通过“show additional metrics”选项进行扩展。

2.2.3.1 stage dag

接下来，我们沿着“stage dag -> event timeline -> task metrics的顺讯，依次讲讲这些页面所包含的内容
首先，我们先来看看最简单的stage dag。点开蓝色的dag visualization按钮，我们就能获取到当前stage的dag，如下图所示。

之所以说stage dag简单，是因为咱们在SQL🎧入口，已经对dag做过详细的说明。而stage dag仅仅是SQL页面完整的dag的子集，毕竟，SQL页面的dag，针对的是作业(job)。因此，只要掌握了作业的dag，自然也就掌握了每一个stage的dag

2.2.3.2 event timeline

与dag visualization并列，在summary metrics之上有一个event timeline的按钮，点开它，我们可以得到如下图所示的可视化信息。
-
even timeline，记录着分布式任务调度与执行的过程中，不同计算环节主要的时间花销。图中的每一个条带，都代表着一个分布式任务，条带由不同颜色构成。其中不同颜色的矩形，代表不同环节的计算时间。

为了方便叙述，我还是同表格形式帮你梳理了这些环节的含义与作用，你可以保存以后随时查看。

metrics	含义
scheduler delay	调度延迟(调度系统开销)
task deserialization time	任务的反序列化时间(调度系统开销)
shuffle read time	shuffle read时间开销
executor computing time	计算时间
shuffle write time	shuffle write时间开销
result serialization time	任务结果的序列化时间
getting result time	结果收集花费的时间

理想情况下，条带的大部分应该都是绿色的，也就是任务的时间消耗，大部分都是执行时间。不过，实际情况并不总是如此，比如，有些时候，蓝色的部分占比较多，或是橙色的部分占比较大
在这些情况下，我们就可以结合event timeline，来判断作业是否存在跳读开销过大，或是shuffle负载过重的问题，从而有针对性的对不同环节做调优。
比方说，如果条带中深蓝的部分(scheduler delay)很多，那就说明任务的调度开销很重。这个时候，我们就需要参考”三足鼎立“的掉调优方法，去相应的调整cpu、内存与并行度、从而减低任务的调度开销。

再比如，如果条带中黄色(shuffle write time)与橙色(shuffle read time)的面积较大，就说明任务的shuffle负载很重，这个时候，我们就需要考虑，有没有可能通过利用broadcast join来消除shuffle，从而缓解任务的shuffle负担。

2.2.3.3 task metrics

说完stage dag和event timeline，最后，我们再来说一说stage详情页的重头戏：task metrics。

之所以说他是重头戏，在于task metrics以不同的粒度，提供了详尽的量化指标。其中，task以task为粒度，记录着每一个分布式任务的执行细节，而summary metrics则是对于所有tasks执行细节的统计汇总。我们先看看粗粒度的summary metrics，载入展开细粒度的task。

2.2.3.4 summary metrics

首先我们点开show additional metrics按钮，勾选select all，让那所有的度量指标都生效，如下图所示。这么做的目的，在于获取最详尽的task执行信息。
可以看到，select all生效之后，spark ui打印出了所有的执行细节。老规矩，为了方便叙述，我还是把这些metrics整理到表格中，方便你随时查阅。其中，task deserialization time、result serialization time、getting result time、scheduler delay与刚刚表格中的含义相同，不再赘述，这里我们紧整理出新出现的task metrics。

metrics	含义
duration	task 执行时间
gc time	任务执行过程中，Java gc时间
peak execution memory	内存消耗峰值
spill(memory)	溢出数据的内存占用
spill(disk)	溢出数据的磁盘占用
shuffle read size/ records	shuffle read 阶段读取的数据量与条目数量
shuffle read blocked time	shuffle read阶段的网络延迟
shuffle remote reads	shuffle read阶段跨节点、从远端节点拉取的数据量
shuffle write size/records	shuffle write阶段写入的数据量与条目数量
shuffle write time	shuffle write阶段话费的写入时间

对于这些详尽的task metrics，难能可贵的，spark UI以最大最小（max min）以及分位点(25%分位、50%分位、75%分位)的方式，提供了不同的metrics的统计分布。这一点非常重要，原因在于，这些metrics的统计分布，可以让我们非常清晰的量化任务的负载分布。
换句话说，根基不同metrics的统计信息分布，我们就可以轻而易举的判定，当前作业的不同任务之间，是相对均衡，还是存在严重的倾斜。如果判断计算存在负载倾斜，那么我们就需要利用手工加盐或者是aqe的自动倾斜处理，去消除任务之间的不均衡，从而改善作业的性能。
在上面的表格中，有一半的metrics是预shuffle直接相关的，比如shuffle read size/record，shuffle remote reads等等。
这些metrics我们在介绍SQL详情的时候，已经详细说过了。另外duration、gctime以及peak execution memory，这些metrics的含义，要么已经讲过，要么过于简单、无需解释。因此，对于这三个指标，咱们也不多写笔墨。
这里要特别值得关注的，是spill（memory）和spill（disk）这两个指标。spill，也即溢出数据，他指的是因内存数据结构（partitionpairbuffer、appendonlymap，等等）空间受限，而腾挪出去的数据。spill（memory）表示的是，这部分数据在内存的存储大小，而spill（disk）表示的是，这些数据在磁盘中的大小。
因此，用spill（memory）除以spill（disk），就可以得到“数据膨胀系数”的近似值，我们把他记为explosion ratio。有了explosion ratio，对于一份存储在磁盘中的数据，我们就可以预估他在内存中的存储大小，从而准确的把我数据的内存消耗。
2.2.3.4 task
- 介绍完粗粒度的summary metrics，接下来，我们再来说说细粒度的tasks。实际上，task的不少指标，是预summary高度重合的，如下图所示。同理，这些重合的metrics，咱们不在赘述，你可以参考summary的部分，来理解这些metrics。唯一的区别，就是这些指标是针对每一个task进行度量的。
  metrics 含义
  locality level 本地性级别
  logs 执行日志
  errors 执行错误细节
  - 可以看到新指标并不多，这里最值得关注的，是locality level，也就是本地性级别。在调度系统重，我们讲过，每个task都有自己的本地性倾向。结合本地性倾向，调度系统会把tasks调度到适度的executors或者是计算节点，尽可能保证数据不动，代码动。
  - logs与errors属于spark ui的三级入口，他们是tasks的执行日志，详细记录了tasks在执行过程中的运行时状态。一般来说，我们不需要深入到三级入口进行debug。errors列提供的报错信息，往往足以让我们迅速的定位问题所在。
  - 思考问题：为什么有跳过的skip stage或者task？
    - skipped stages表示已经执行过了。
      rdd rdd.filter.map => transformation
      rdd.cache
      count reduce的操作两个都是action
      
      count
      执行reduce的时候不需要重新计算他的transformation
      
      3 Spark 参数调优与Spark sql调优
      
      3.1 spark数据倾斜的解决方案
      
      产生数据倾斜的主要原因是在shuffle中不同的key对应的数据量不同，导致不同的task分配的数据量不均衡。
      - 提高shuffle的并行度
      - 使用随机数前缀。将相同的key增加不同的前缀，使这些相同的key分散到不同的task进行处理。
        缺点：对内存要求高
        reduce join转为map join。适用于两个表里面一个表比较小的场景，在map端进行小表广播，用map算子实现与join相同的效果。
        优点：不需要shuffle
        缺点：只只用于大表join小表的情况。
        过滤少数导致数据倾斜的key
        缺点：数据量多的key没有业务使用上的含义。场景单一
        优点：实现简单。
        在hive进行预处理，然后将数据传给hive
        使用两阶段预聚合操作。
        先局部聚合，再做全局聚合。适用于reducebykey groupby的场景。
        优点：显著提升spark性能
        缺点：适用于最固定的场景

metrics	含义
locality level	本地性级别
logs	执行日志
errors	执行错误细节

分类:热门推荐日期:2024-04-29浏览:1评论:0