SparkSQL-性能调优

news/2024/10/2 8:01:29 标签: spark, 大数据, 分布式, scala, sql

祝福

在这个举国同庆的时刻,我们首先献上对祖国的祝福:

第一,我们感谢您给我们和平的环境,让我们能快乐生活

第二,祝福我们国家未来的路越走越宽广,科技更发达,人民更幸福

第三,我们会紧紧跟随您的脚步,一起为美好的未来奋斗

一、将数据缓存到内存

import spark.implicits._
import spark.sql

spark.catalog.cacheTable("personal_info")
spark.catalog.uncacheTable("personal_info")
val personalInfoDataFrame = sql("select name,age FROM personal_info")
personalInfoDataFrame.cache()
personalInfoDataFrame.unpersist()

cacheTable() 和 cache() 最终都调用了CacheManager的cacheQuery()

CacheManager对数据的缓存与RDD中的cache()是不同的,RDD中的cache()是的持久化级别是MEMORY_ONLY,而这里是MEMORY_AND_DISK,因为Spark任务重新计算底层表并将其缓存是昂贵的。

二、选项设置

以下选项可用于调整查询执行的性能。随着自动执行更多优化,这些选项可能会在未来的版本中被弃用。

spark.sql.files.maxPartitionBytes        默认:128M

解释:读取文件时打包到单个分区中的最大字节数。此配置仅在使用基于文件的源(如Parket、JSON和ORC)时有效。

spark.sql.files.openCostInBytes        默认值:4M

解释:打开一个文件的估计成本,以可以同时扫描的字节数来衡量。这是在将多个文件放入一个分区时使用的。最好过度估计,那么具有小文件的分区将比具有较大文件的分区(这是首先调度的)更快。此配置仅在使用基于文件的源(如Parket、JSON和ORC)时有效。

spark.sql.files.minPartitionNum        默认值:spark.sql.leafNodeDefaultParallelism

解释:建议的(不保证的)最小分割文件分区数如果未设置。此配置仅在使用基于文件的源(如Parquet、JSON和ORC)时有效。

spark.sql.files.maxPartitionNum        默认值:None

解释:建议的(不保证的)分割文件分区的最大数量。如果设置了,如果初始分区数超过此值,Spark将重新调整每个分区以使分区数接近此值。此配置仅在使用基于文件的源(如Parket、JSON和ORC)时有效。

spark.sql.broadcastTimeout        默认值:5*60 秒

解释:广播等待超时时间

spark.sql.autoBroadcastJoinThreshold        默认值:10M

解释:配置表的最大大小(以字节为单位),该表将在以下情况下广播到所有工作节点 执行连接。通过将此值设置为-1,可以禁用广播。请注意,目前 统计信息仅支持Hive Metastore表,其中命令 为:

ANALYZE TABLE <tableName> COMPUTE STATISTICS noscan (用于收集表的统计信息,这些统计信息将帮助优化器制定更好的执行计划

也就是Spark3.2.0以后不需要自己去优化大小表join了,比如手动广播小表、大小表位置

spark.sql.shuffle.partitions        默认值:200

解释:配置为连接或聚合混洗数据时要使用的分区数。在入库时如果最终的结果不是很大,需求将其调小,不然入库的时间会增大。

spark.sql.sources.parallelPartitionDiscovery.threshold        默认值:32

解释:配置阈值以启用作业输入路径的并行列表。如果输入路径的数量大于此阈值,Spark将使用Spark分布式作业列出文件。否则,它将回退到顺序列表。此配置仅在使用基于文件的数据源(如Parquet、ORC和JSON)时有效。

spark.sql.sources.parallelPartitionDiscovery.parallelism        默认值:10000

解释:配置作业输入路径的最大列表并行性。如果输入路径的数量大于此值,则将限制其使用此值。此配置仅在使用基于文件的数据源(如Parquet、ORC和JSON)时有效。

三、Hint 干预

Hint 是一种可以让用户干预数据库 SQL 优化的方式,相当于给用户开了一个后门,当优化器本身对于某些 SQL 优化得不够好时,用户就可以结合自己的经验,尝试使用 Hint 来干预数据库的优化。

1、join 的 Hint

sql语句中使用BROADCAST、MERGE、SHUFLE_HASH和SHUFLE_REPLICATE_NL,指示Spark在将它们与另一个关系连接时,对每个指定的关系使用提示策略。例如,当在表“t1”上使用BROADCAST提示时,Spark将优先考虑以“t1”作为构建端的广播连接(广播哈希连接或广播嵌套循环连接,具体取决于是否有任何equi-join键),即使统计数据建议的表“t1“的大小高于配置Spark.sql.autoBroadcastJoinThreshold。

当在连接的两侧指定不同的连接策略提示时,Spark会将BROADCAST提示优先于MERGE提示,将SHUFLE_HASH提示优先于SHUFLE_REPLICATE_NL提示。当使用BROADCAST提示或SHUFLE_HASH提示指定两侧时,Spark将根据连接类型和关系大小选择构建侧。

请注意,不能保证Spark会选择提示中指定的连接策略,因为特定的策略可能不支持所有连接类型。

示例:

spark.table("src").join(spark.table("records").hint("broadcast"), "key").show()
sql">-- We accept BROADCAST, BROADCASTJOIN and MAPJOIN for broadcast hint
SELECT /*+ BROADCAST(r) */ * FROM records r JOIN src s ON r.key = s.key

2、Coalesce 的 Hint

Coalesce 的 Hint 允许Spark SQL用户控制输出文件的数量,就像Dataset API,中的coalesce、repartition和repartitionByRange一样,它们可以用于性能调整和减少输出文件的数目。

sql">SELECT /*+ COALESCE(3) */ * FROM t;
SELECT /*+ REPARTITION(3) */ * FROM t;
SELECT /*+ REPARTITION(c) */ * FROM t;
SELECT /*+ REPARTITION(3, c) */ * FROM t;
SELECT /*+ REPARTITION */ * FROM t;
SELECT /*+ REPARTITION_BY_RANGE(c) */ * FROM t;
SELECT /*+ REPARTITION_BY_RANGE(3, c) */ * FROM t;
SELECT /*+ REBALANCE */ * FROM t;
SELECT /*+ REBALANCE(3) */ * FROM t;
SELECT /*+ REBALANCE(c) */ * FROM t;
SELECT /*+ REBALANCE(3, c) */ * FROM t;

四、自适应查询AQE

AQE是Adaptive Query Execution的简称,是Spark SQL中的一种优化技术,它利用运行时统计信息来选择最有效的查询执行计划,自Apache Spark 3.2.0以来,该计划默认启用。Spark SQL可以通过作为伞形配置启用Spark.SQL.adaptive.enabled来打开和关闭AQE。从Spark 3.0开始,AQE中有三个主要功能:包括合并洗牌后分区、将排序合并连接转换为广播连接,以及倾斜连接优化。

1、shuffle分区自动调整

这里涉及两个配置:

spark.sql.adaptive.enabled        默认:true

解释:如果为true,则启用自适应查询执行,该操作将在查询执行过程中根据准确的运行时统计信息重新优化查询计划

spark.sql.adaptive.coalescePartitions.enabled        默认:true

解释:当true和spark.sql.adaptive.enabled为true时,Spark将根据目标大小(由spark.sql.adaptive.advisoryPartitionSizeInBytes指定)合并连续的shuffle分区,以避免过多的小任务

其他配置:

spark.sql.adaptive.coalescePartitions.parallelismFirst        默认:true

解释:如果为true,则Spark在合并连续的shuffle分区时忽略spark.sql.adaptive.advisoryPartitionSizeInBytes指定的目标大小(默认64MB),并且只尊重由spark.sql.adaptive.coalescePartitions.minPartitionSize指定的最小分区大小(默认1MB),以最大化并行性。这是为了避免启用自适应查询执行时性能下降。建议将此配置设置为false,并尊重由spark.sql.adaptive.advisoryPartitionSizeInBytes指定的目标大小。

spark.sql.adaptive.coalescePartitions.minPartitionSize     默认:1M

解释:合并后shuffle分区的最小大小。它的值最多可以是20%spark.sql.adaptive.advisoryPartitionSizeInBytes。当在分区合并期间忽略目标大小时,这很有用,这是默认情况。

spark.sql.adaptive.coalescePartitions.initialPartitionNum        默认:none

解释:合并前shuffle分区的初始数量。如果未设置,则等于spark.sql.shuffle.partitions。此配置仅在spark.sql.adaptive.enabledspark.sql.adaptive.coalescePartitions.enabled时有效

spark.sql.adaptive.advisoryPartitionSizeInBytes        默认:64M

解释:自适应优化期间shuffle分区的建议大小(以字节为单位spark.sql.adaptive.enabled

该功能简化了运行查询时shuffle分区号的调整。用户不需要设置适当的shuffle分区号来适应数据集。一旦你通过Spark.sql.adaptive.coalescePartitions.initialPartitionNum配置设置了足够大的初始洗牌分区数,Spark就可以在运行时选择合适的洗牌分区数。

以下是对倾斜分区的相关设置

spark.sql.adaptive.optimizeSkewsInRebalancePartitions.enabled         默认:true

解释:当spark.sql.adaptive.enabled为true时,Spark将优化Rebalance分区中的倾斜shuffle分区,并根据目标大小(由spark.sql.adaptive.advisoryPartitionSizeInBytes指定)将其拆分为较小的分区,以避免数据倾斜。

spark.sql.adaptive.rebalancePartitionsSmallPartitionFactor         默认:0.2

解释:如果分区的大小小于此因数,则分区将在拆分期间合并spark.sql.adaptive.advisoryPartitionSizeInBytes

2、将sort-merge join转换为broadcast join

当任何连接端的运行时统计数据小于自适应广播哈希连接阈值时,AQE将排序合并连接转换为广播哈希连接。这不如一开始就计划广播哈希连接有效,但它比继续进行排序合并连接要好,因为我们可以保存连接双方的排序,并在本地读取shuffle文件以节省网络流量(如果spark.sql.adaptive.localShuffleReader.enabled为真)

spark.sql.adaptive.autoBroadcastJoinThreshold        默认:none

解释:为执行连接时将广播到所有工作节点的表配置最大大小(以字节为单位)。通过将此值设置为-1,可以禁用广播。默认值与spark.sql.autoBroadcastJoinThreshold相同。请注意,此配置仅在自适应框架中使用。

spark.sql.adaptive.localShuffleReader.enabled        默认:true

解释:当true且spark.sql.adaptive.enabled为true时,Spark会在不需要shuffle分区时尝试使用本地shuffle阅读器读取shuffle数据,例如,在将sort-合并连接转换为广播散列连接之后。

AQE将排序合并连接转换为洗牌哈希连接当所有后洗牌分区都小于阈值时,最大阈值可以看到配置spark.sql.adaptive.maxShuffledHashJoinLocalMapThreshold

spark.sql.adaptive.maxShuffledHashJoinLocalMapThreshold        默认:0

解释:配置每个分区可以允许构建本地哈希映射的最大大小(以字节为单位)。如果此值不小于spark.sql.adaptive.advisoryPartitionSizeInBytes并且所有分区大小都不大于此配置,则连接选择更喜欢使用洗牌哈希连接而不是排序合并连接,而不管spark.sql.join.preferSortMergeJoin的值如何

3、倾斜join优化

数据倾斜会严重降低连接查询的性能。此功能通过将倾斜的任务拆分(并在需要时复制)成大致均匀大小的任务来动态处理排序合并连接中的倾斜。当spark.sql.adaptive.enabledspark.sql.adaptive.skewJoin.enabled配置都启用时,它会生效。

spark.sql.adaptive.skewJoin.enabled        默认:true

解释:当true和spark.sql.adaptive.enabled为true时,Spark通过拆分(并在需要时复制)倾斜的分区来动态处理排序合并连接中的倾斜。

spark.sql.adaptive.skewJoin.skewedPartitionFactor         默认:5.0

解释:如果一个分区的大小大于乘以中值分区大小的因子,并且也大于spark.sql.adaptive.skewJoin.skewedPartitionThresholdInBytes,则该分区被认为是倾斜的。

spark.sql.adaptive.skewJoin.skewedPartitionThresholdInBytes        默认:256MB

解释:如果分区的大小(以字节为单位)大于此阈值,并且大于spark.sql.adaptive.skewJoin.skewedPartitionFactor乘以分区大小的中值,则该分区被视为倾斜。理想情况下,此配置应设置为大于spark.sql.adaptive.advisoryPartitionSizeInBytes

spark.sql.adaptive.forceOptimizeSkewedJoin         默认:false

解释:如果为true,则强制启用OptimizeSkewedJoin,这是一种自适应规则,用于优化偏斜连接以避免散乱任务,即使它引入了额外的洗牌

 


http://www.niftyadmin.cn/n/5688176.html

相关文章

【论文阅读】基于真实数据感知的模型功能窃取攻击

摘要 目的 模型功能窃取攻击是人工智能安全领域的核心问题之一&#xff0c;目的是利用有限的与目标模型有关的信息训练出性能接近的克隆模型&#xff0c;从而实现模型的功能窃取。针对此类问题&#xff0c;一类经典的工作是基于生成模型的方法&#xff0c;这类方法利用生成器…

Java面向对象第四章方法重写与多态练习题

练习1&#xff1a;使用重写优化电子宠物系统 需求说明 使用方法重写优化电子宠物系统&#xff0c;实现如下效果 package com.hz.ch02; /*** 猫类* author 26255**/ public class Cat extends Father {private String sex;public Cat(String name, int health, int love, Stri…

Python面向对象基础:属性动态添加

目录 前言 一、面向对象编程概述 1.1 什么是面向对象编程? 1.2 Python中的OOP 二、类和对象的基本定义 2.1 定义类 2.2 创建对象 三、动态添加属性 3.1 如何动态添加属性 3.2 示例:动态添加属性 3.3 动态添加方法 四、动态添加属性的应用场景 4.1 配置对象 4.2 …

Android 13.0 系统wifi列表显示已连接但无法访问网络问题解决

1.前言 在13.0的系统rom产品定制化开发中,在wifi模块也很重要,但是在某些情况下对于一些wifi连接成功后,确显示已连接成功,但是无法访问互联网 的情况,所以实际上这时可以正常上网的,就是显示的不正常,所以就需要分析连接流程然后解决问题 如图所示: 2.系统wifi列表显示…

Elasticsearch基础_3.基础操作

文章目录 一、索引操作1.1、创建索引1.2、删除索引 二、映射操作2.1、查看映射2.2、扩展映射 三、文档操作3.1、单条写入文档3.2、更新单条文档3.3、查看单条文档3.4、删除单条文档3.5、根据条件删除文档 一、索引操作 1.1、创建索引 PUT /${index_name} {"settings&quo…

vim/vi常用命令大全

启动和退出Vim 命令/操作作用vim启动Vimvim filename直接打开指定的文件命令模式下&#xff0c;输入 :q退出&#xff0c;q!强制退出:wq保存并退出:wq!保存并强制退出vim中按下a进入编辑模式Esc退出编辑模式进入命令模式new创建新窗口close关闭窗口 光标移动 命令/操作作用h、…

前端——Ajax和jQuery

一、Ajax Ajax即“Asynchronous Javascript And XML”&#xff08;异步 JavaScript 和 XML&#xff09;&#xff0c; 通过 JS 异步的向服务器发送请 求并接收响应数据。 同步访问&#xff1a;当客户端向服务器发送请求时&#xff0c;服务器在处理的过程中&#xff0c;浏览器…

怎样过好国庆节

今天是2024年10月1号&#xff0c;国庆节&#xff0c;七天小长假&#xff0c;估计每个人都有自己的小计划。有想出去浪的&#xff0c;有想闭关修炼的&#xff0c;有想约会恋爱的&#xff0c;也有想回家看父母的&#xff0c;只要有事干&#xff0c;有想法&#xff0c;有行动&…