技术资讯 | Spark sql优化案例分享

01

引言

首先我们从 两种计算引擎的基本架构图来分析Spark/MapReduce的性能 。

1.1 M apReduce

 

处理效率低效 :

• Map/Reduce任务中间结果写磁盘，多个MR之间通过HDFS交换数据; 任务调度和启动开销大；

• 一条 SQL 语句经常被拆分成多个Application，数据在多个Application之间只能通过读写HDFS交换；

• 无法充分利用内存;

1.2 Spark

1.2.1高 效 (比MapReduce快几倍到几十倍) 

• 内存计算引擎，提供Cache机制来支持需要反复迭代计算或者多次数据共享，减少数据读取的IO开销，另外为了解决纯内存计算带来的数据可靠性，引入了Checkpoint机制;

• DAG引擎，减少多次计算之间中间结果写到HDFS的开销;

• Executor使用线程池模型来减少task启动开销，shuffle过程中避免 不必要的sort操作以及减少磁盘IO操作。

02

影响Spark任务快慢的因素

2.1因子

• 数据量 (GB级 vs TB级)

• 数据组织形式 (存储结构，压缩算法，数据Schema[Map等复杂结构])

• 小文件 (很多的KB级，10M级)

• 内存 (内存偏少，大量溢写)

• Core个数 (并发量小，shuffle等待)

• 算法 (聚合因子，过滤条件，SQL组织形式 )

• 发生大量的Shuffle (可用Broadcast替换)

• 是否使用缓存 (将经常访问的数据缓存在Executor内存中) 

2.2.Spark基本流程图

 

• Driver (AppMaster) 负责DAG,Task调度

• Yarn负责计算资源分配

• Worker上的Executor负责计算

Worker节点才是计算执行的地方，所以性能优化重点将研究Executor的优化。

2.3Executor

建议不要自己配置Executor个数，使用动态分配模式:

概念：

根据当前的负载动态的增加或者删除Executor，这样做的 好处 在于：

在业务组的队列资源 (vcore, memory) 资源恒定额情况下，能更好的均衡各个业务的对资源的占用，也就是对于一个计算量较小的任务不用占用太多资源。而对于一个计算量较大的任务，也能从集群中获取相对较多的资源。

而采用指定模式，则会导致任务在获取足够多 (可通过参数设置比例) 的Executor之前一直处于等待状态，而这通常会浪费计算资源。

Executor动态分配模型

 

ExecutorAllocationManager内部会定时根据工作负载计算所需的Executor数量:

• 如果任务对Executor需求数量大于之前向集群管理器申请的Executor数量，那么向Yarn申请添加Executor;

• 如果任务对Executor需求数量小于之前向集群管理器申请的Executor数量，那么向Yarn申请取消部分Executor;

• ExecutorAllocationManager内部还会定时向Yarn申请移除 (杀死) 过期的Executor;

2.4Core

spark.executor.cores

建议：

• 建议executor的cpu core数量设置为2 ~ 4个比较合适；

• 在队列有大量任务提交的情况下，还要更少，以免影响其他用户提交的任务因申请不到cpu (vcore) 资源而卡住。

2.5Memory

spark.executor.memory

建议：

每个Executor的每个core分配的内存设置4g较为合适。

用户设置该值的时候需要考虑如下影响因子：

Spark内存区分存储内存&计算内存

 

spark.storage.memoryFraction

控制存储内存在整个内存中的比例

StorageMemory：

RDD cache, RDD Broadcast等等内容

根据应用的不同可自己动态调整，但通常情况下不需要调整。使用默认值即可，下图展示的是Storage内存与Execution的 内存动态调节机制 。

 

2.6 Shuffle并行度

spark.default.parallelism

该参数用于设置每个stage的默认task数量，这个参数极为重要，如果不设置可能会直接影响你的任务性能。 (只有在处理RDD时才会起作用，对Spark SQL无效)

建议：

500 ~ 1000较为合适，通常情况设置为executor-memory * num-executor 乘积的2~3倍较为合适；

spark.sql.shuffle.partitions

用于配置join 或聚合操作shuffle数据时使用的分区数 (对sparks SQL专用的设置)

2.7 存储结构

目前HADOOP中常用的数据存储结构包括：

• Text (行式存储)

• CSV (行式存储)

• RCFile (列式存储)

• ORC (列式存储)

• Parquet (列式存储)

目前Spark默认存储的格式为Parquet。 下图展示的是相同数据以不同存储结构存储，存储文件的Size对比：

 

2.8 列式存储的好处

• 查询的时候不需要扫描全部的数据，而只需要读取每次查询涉及的列。 这样可以将I/O消耗降低N倍，另外可以保存每一列的统计信息 (min 、 max、sum等) ，实现部分的谓词下推;

• 由于每一列的成员都是同构的，可以针对不同的数据类型使用更高效的数据压缩算法，进一步减小I/O;

• 由于每一列的成员的同构性，可以使用更加适合CPU pipeline的编码方式，减小CPU的缓存失效;

• 由于列式存储数据量更小，Spark的Task读取数据的时间更短，不光节省计算资源，还节省存储资源。

 

列式存储的向量化操作，相对于行式存储一行一行的操作，列式存储可做到 一个batch的操作，这样的操作方式极大的 提升了运算性能 。

 

2.9 压缩方式(gzip, bzip2, lzo, snappy)

案例1：

Snappy压缩前后比例为3:1

 

对于Spark任务来说，压缩的数据带来的好处是显而易见的：

• 大幅节省内存

• 大幅节省磁盘

• 大幅节省数据读取时间

各种压缩格式特性

 

Spark设置压缩格式 (例如设置snappy压缩) ：

各种压缩格式性能对比

 

2.10 输入小文件

案例1：

 

• 小文件太多，导致每个task读取的数据量较小，计算的时间很短；

• 执行的时间不足以弥补JVM启动的时间。

调节每个task任务的输入数据大小

2.11 Shuffle&&输出小文件

某个表的hdfs文件如下：

输出表的分区下有20个小文件。

由于集群中NameNode节点需要维护文件的元数据信息，太多的输出小文件会给集群的NameNode带来巨大的压力；

控制task个数

方案1：

使用hint将会使得输入数据进行重新Repartition，调节最终task的个数以及输出文件的个数。

通过重分区将减少或者增大分区数量以达到增加或减少task的数量，从而增大或者减少Task输出的文件个数。

2.12 SQL自身-下推 (PushDownPredicate)

 

上图中：

• 方式1从磁盘中读取出所有的数据，在内存中过滤；

• 方式2,3将过滤从内存中下推到磁盘，在扫描磁盘的数据的时候就过滤掉数据。

概念：

所谓下推是指将过滤尽可能地下沉到数据源端，从而避免从磁盘读取不必要数据。

下推与不下推性能对比 (DataBricks官方)

 

通过上图可知，60%的下推比不下推的性能提高了2~18倍；

2.12.1谓词下推的限制：

• 只有operator 包含的所有expression都是确定性的时候才可以下推， 比如 rand 表达式等;

• Filter 的字段必须要在group by 的维度字段里面，举个例子：

1）下面的聚合是 可以谓词下推 的：

2）下面的聚合是 不可以谓词下 推的：

案例 1：

Partition是String类型，此处是Int类型，导致下推失败。

 

案例 2：

Regexp是非确定的，导致下推失败

 

2.13 SQL自身-广播

Spark：

对该变量进行广播。广播后的变量，会保证每个Executor的内存中，只驻留一份变量副本，而Executor中的task执行时共享该Executor中的那份变量副本。

这样的话，可以大大减少变量副本的数量，从而减少网络传输的性能开销，并减少对Executor内存的占用开销，降低GC的频率；

可适当调整广播变量大小的阈值，使得稍微大一些的数据也能被广播：

 

2.14 SQL自身-缓存

将数据缓存在内存中，需要 遵循的原则为：

• 数据重复使用

• 重新生成这部分数据的代价昂贵

SQL语句：

权衡cache与否的代价，不cache则多次使用同一份数据都需要重新计算一次。

Cache则只会计算一次，但是会占用executor的内存资源，那是否应该cache就是把计算RDD,从hdfs上获取数据的时间资源与缓存数据的内存资源之间进行权衡。

 

  rdd1,rdd2不需要缓存

rdd可以缓存，rdd1,rdd2不需要缓存

2.15 表结构嵌套字段(Map,Array )

案例1：

 

这张表是业务用户的表结构。

用户行为数据以Json形式上报，由于表的结构实在太过复杂。Column字段存在大量的Map结构，分析层面很难通过简单的SQL语句来分析这行数据，只能以读取HDFS,在代码层面来做数据分析。

案例2：

Spark SQL 处理嵌套类型数据时，存在以下问题：

1）读取大量不必要的数据：

对于嵌套数据类型的字段，如下图中的Map 类型的people 字段，往往只需要读取其中的子字段，如people.age。

却需要将整个Map 类型的people 字段全部读取出来然后抽取出people.age 字段。这会引入大量的无意义的IO 开销。如果是几百个Key，这也就意味着IO 被放大了几十至几百倍；

2）无法进行向量化读取：

而向量化读能极大的提升性能。Spark 不支持包含嵌套数据类型的向量化读取，这极大地影响了包含嵌套数据类型的查询性能；

3）不支持 Filter 下推：

Spark 不支持嵌套类型字段上的Filter 的下推；

4）重复计算：

JSON 字段，在Spark SQL 中以String 类型存在，严格来说不算嵌套数据类型。不过实践中也常用于保存不固定的多个字段，在查询时通过JSON Path 抽取目标子字段。

而大型JSON 字符串的字段抽取非常消耗CPU。对于热点表，频繁重复抽取相同子字段非常浪费资源。

2.16 Distribute By / Cluster By

好处：

在Join的过程中，未使用Distribute By 的两个表，将各自的数据随机分配到executor上，当join的时候就会出现，跨executor的数据需要互相迁移进行匹配的情况，从而引起更大量的Shuffle。

但如果我们将join的连接字段通过Distribute By 重分区，神奇的一幕将发生：需要互相匹配的数据都会在相同的executor上，从而避免跨executor的数据迁移，所有的匹配都在统一个executor上进行，这样就大大减少了Shuffle。

 

使用方法：

 

投稿 | 大数据平台

编辑 | sea

排版 | sea

往期推荐

在看点一下 大家都知道