pyspark系列10-Spark SQL性能调优常用参数介绍

秒速五厘米 2023-01-20 13:58 13阅读 0赞

### 文章目录 ###

*  一.在内存中缓存数据
 *  二.其它配置项
 *  三.SQL查询连接的hint
 *  四.自适应查询执行
 *  五.合并分区后重新组合
 *  六.将排序合并联接转换为广播联接
 *  七.优化倾斜连接
 *  参考:

# 一.在内存中缓存数据 #

Spark SQL可以通过调用Spark.catalog.cachetable (“tableName”)或DataFrame.cache()来使用内存中的columnar格式缓存表。然后Spark SQL将只扫描所需的列，并自动调优压缩以最小化内存使用和GC压力。你可以调用spark.catalog.uncacheTable(“tableName”)从内存中删除表。

内存缓存的配置可以在SparkSession上使用setConf方法或者使用SQL运行SET key=value命令来完成。

<table> 
 <thead> 
  <tr> 
   <th>参数名</th> 
   <th>默认值</th> 
   <th>参数说明</th> 
   <th>启始版本</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>spark.sql.inMemoryColumnarStorage.compressed</td> 
   <td>true</td> 
   <td>当设置为true时，Spark SQL会根据数据统计自动为每列选择压缩编解码器。</td> 
   <td>1.0.1</td> 
  </tr> 
  <tr> 
   <td>spark.sql.inMemoryColumnarStorage.batchSize</td> 
   <td>10000</td> 
   <td>控制柱状缓存的批大小。更大的批处理大小可以提高内存利用率和压缩，但在缓存数据时可能会带来OOMs风险。</td> 
   <td>1.1.1</td> 
  </tr> 
 </tbody> 
</table>

# 二.其它配置项 #

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

<table> 
 <thead> 
  <tr> 
   <th>参数名</th> 
   <th>默认值</th> 
   <th>参数说明</th> 
   <th>启始版本</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>spark.sql.files.maxPartitionBytes</td> 
   <td>134217728 (128 MB)</td> 
   <td>读取文件时装入单个分区的最大字节数。此配置仅在使用基于文件的源(如Parquet、JSON和ORC)时有效。</td> 
   <td>2.0.0</td> 
  </tr> 
  <tr> 
   <td>spark.sql.files.openCostInBytes</td> 
   <td>4194304 (4 MB)</td> 
   <td>打开一个文件的估计成本，由可以在同一时间扫描的字节数来衡量。当将多个文件放入一个分区时使用。最好是高估，那么带有小文件的分区将比带有大文件的分区更快(这是首先安排的)。此配置仅在使用基于文件的源(如Parquet、JSON和ORC)时有效。</td> 
   <td>2.0.0</td> 
  </tr> 
  <tr> 
   <td>spark.sql.files.minPartitionNum</td> 
   <td>Default Parallelism</td> 
   <td>建议的(不是保证的)最小分割文件分区数。如果没有设置，默认值是’ spark.default.parallelism '。此配置仅在使用基于文件的源(如Parquet、JSON和ORC)时有效。</td> 
   <td>3.1.0</td> 
  </tr> 
  <tr> 
   <td>spark.sql.broadcastTimeout</td> 
   <td>300</td> 
   <td>broadcast join 等待时间的超时(秒)</td> 
   <td>1.3.0</td> 
  </tr> 
  <tr> 
   <td>spark.sql.autoBroadcastJoinThreshold</td> 
   <td>10485760 (10 MB)</td> 
   <td>配置在执行联接时将广播到所有工作节点的表的最大字节大小。通过将此值设置为-1，可以禁用广播。注意:目前统计只支持运行ANALYZE TABLE COMPUTE statistics noscan命令的Hive Metastore表。</td> 
   <td>1.1.0</td> 
  </tr> 
  <tr> 
   <td>spark.sql.shuffle.partitions</td> 
   <td>200</td> 
   <td>配置将数据变换为连接或聚合时要使用的分区数量。</td> 
   <td>1.1.0</td> 
  </tr> 
  <tr> 
   <td>spark.sql.sources.parallelPartitionDiscovery.threshold</td> 
   <td>32</td> 
   <td>配置阈值以启用作业输入路径的并行列出。如果输入路径数大于该阈值，Spark将通过Spark分布式作业列出文件。否则，它将退回到顺序列表。此配置仅在使用基于文件的数据源(如Parquet、ORC和JSON)时有效。</td> 
   <td>1.5.0</td> 
  </tr> 
  <tr> 
   <td>spark.sql.sources.parallelPartitionDiscovery.parallelism</td> 
   <td>10000</td> 
   <td>配置作业输入路径的最大列出并行度。如果输入路径的数量大于这个值，它将被降低到使用这个值。与上面一样，此配置仅在使用基于文件的数据源(如Parquet、ORC和JSON)时有效。</td> 
   <td>2.1.1</td> 
  </tr> 
 </tbody> 
</table>

# 三.SQL查询连接的hint #

join策略提示BROADCAST、MERGE、SHUFFLE\_HASH和SHUFFLE\_REPLICATE\_NL，在将指定的关系加入到另一个关系时，指示Spark对每个指定的关系使用暗示策略。例如，在表’ t1 '上使用BROADCAST提示时，广播加入(广播散列连接或广播嵌套循环联接取决于是否有等值连接键)与t1的构建方面将由火花即使大小的优先表t1的建议的统计配置spark.sql.autoBroadcastJoinThreshold之上。

当连接两端指定了不同的连接策略提示时，Spark会优先考虑BROADCAST提示而不是MERGE提示，优先考虑SHUFFLE\_HASH提示而不是SHUFFLE\_REPLICATE\_NL提示。当双方都指定了BROADCAST提示或SHUFFLE\_HASH提示时，Spark将根据连接类型和关系的大小选择构建端。

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

-- We accept BROADCAST, BROADCASTJOIN and MAPJOIN for broadcast hint
    SELECT /*+ BROADCAST(r) */ * FROM records r JOIN src s ON r.key = s.key

Coalesce hint允许Spark SQL用户控制输出文件的数量，就像Dataset API中的Coalesce、repartition和repartitionByRange一样，它们可以用于性能调优和减少输出文件的数量。COALESCE hint只有一个分区号作为参数。“REPARTITION”提示有一个分区号、列或它们都作为参数。“REPARTITION\_BY\_RANGE”提示必须有列名，分区号是可选的。

SELECT /*+ COALESCE(3) */ * FROM t
    SELECT /*+ REPARTITION(3) */ * FROM t
    SELECT /*+ REPARTITION(c) */ * FROM t
    SELECT /*+ REPARTITION(3, c) */ * FROM t
    SELECT /*+ REPARTITION_BY_RANGE(c) */ * FROM t
    SELECT /*+ REPARTITION_BY_RANGE(3, c) */ * FROM t

# 四.自适应查询执行 #

Adaptive Query Execution (AQE)是Spark SQL中的一种优化技术，它利用运行时统计信息来选择最高效的查询执行计划。默认情况下AQE是禁用的。Spark SQL可以使用Spark.SQL.adaptive.enabled的伞配置来控制是否打开/关闭。从Spark 3.0开始，AQE中有三个主要特性，包括合并shuffle后分区、将排序合并连接转换为广播连接以及倾斜连接优化。

# 五.合并分区后重新组合 #

当spark.sql.adaptive.enabled和spark.sql.adaptive.coalescePartitions.enabled配置都为true时，该特性根据map输出统计信息来合并post shuffle分区。这个特性简化了运行查询时shuffle分区号的调优。您不需要设置合适的shuffle分区号来适合您的数据集。一旦您通过Spark .sql. adaptive.coalescepartitions . initialpartitionnum配置设置了足够大的初始shuffle分区数，Spark就可以在运行时选择适当的shuffle分区号。

<table> 
 <thead> 
  <tr> 
   <th>参数名</th> 
   <th>默认值</th> 
   <th>参数说明</th> 
   <th>启始版本</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>spark.sql.adaptive.coalescePartitions.enabled</td> 
   <td>true</td> 
   <td>当true和Spark .sql. adaptive_enabled为true时，Spark会根据目标大小(由Spark .sql. adaptive_advisorypartitionsizeinbytes指定)合并连续的shuffle分区，以避免过多的小任务。</td> 
   <td>3.0.0</td> 
  </tr> 
  <tr> 
   <td>spark.sql.adaptive.coalescePartitions.minPartitionNum</td> 
   <td>Default Parallelism</td> 
   <td>合并后的最小洗牌分区数。如果不设置，则默认为Spark集群的默认并行度。此配置仅在spark.sql. adaptive.net enabled和spark.sql. adaptive.net coalescepartitions .enabled同时启用时有效。</td> 
   <td>3.0.0</td> 
  </tr> 
  <tr> 
   <td>spark.sql.adaptive.coalescePartitions.initialPartitionNum</td> 
   <td>200</td> 
   <td>合并前的初始shuffle分区数。默认情况下它等于spark.sql.shuffle.partitions。此配置仅在spark.sql. adaptive.net enabled和spark.sql. adaptive.net coalescepartitions .enabled同时启用时有效。</td> 
   <td>3.0.0</td> 
  </tr> 
  <tr> 
   <td>spark.sql.adaptive.advisoryPartitionSizeInBytes</td> 
   <td>64 MB</td> 
   <td>自适应优化期间shuffle分区的建议大小(当spark.sql. adaptive_enabled为true时)。当Spark对小shuffle分区或斜shuffle分区进行合并时生效。</td> 
   <td>3.0.0</td> 
  </tr> 
 </tbody> 
</table>

# 六.将排序合并联接转换为广播联接 #

当任何连接侧的运行时统计数据小于广播散列连接阈值时，AQE将排序合并连接转换为广播散列连接。这不是一样有效规划一个广播散列连接首先,但这总比继续做分类合并加入,我们可以节省连接双方的排序,并在本地读取洗牌文件节省网络流量(如果spark.sql.adaptive.localShuffleReader.enabled被设置为true)

# 七.优化倾斜连接 #

数据倾斜会严重降低连接查询的性能。该特性通过将倾斜任务拆分(如果需要的话还可以复制)为大小大致相同的任务，动态处理排序-合并连接中的倾斜任务。当spark.sql.adaptive.enabled和spark.sql.adaptive.skewJoin.enabled配置同时启用时生效。

<table> 
 <thead> 
  <tr> 
   <th>参数名</th> 
   <th>默认值</th> 
   <th>参数说明</th> 
   <th>启始版本</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>spark.sql.adaptive.skewJoin.enabled</td> 
   <td>true</td> 
   <td>当true和Spark .sql.adaptive.enabled为true时，Spark通过拆分(并在需要时复制)倾斜分区来动态处理排序-合并连接中的倾斜。</td> 
   <td>3.0.0</td> 
  </tr> 
  <tr> 
   <td>spark.sql.adaptive.skewJoin.skewedPartitionFactor</td> 
   <td>10</td> 
   <td>如果一个分区的大小大于这个因子乘以中值分区大小，并且大于spark.sql.adaptive.skewJoin.skewedPartitionThresholdInBytes，则认为该分区是倾斜的。</td> 
   <td>3.0.0</td> 
  </tr> 
  <tr> 
   <td>spark.sql.adaptive.skewJoin.skewedPartitionThresholdInBytes</td> 
   <td>256MB</td> 
   <td>如果分区的字节大小大于这个阈值，并且大于spark.sql.adaptive.skewJoin.skewedPartitionFactor乘以分区中值大小，则认为该分区是倾斜的。理想情况下，该配置应该设置为大于spark.sql.adaptive.advisoryPartitionSizeInBytes。</td> 
   <td>3.0.0</td> 
  </tr> 
 </tbody> 
</table>

# 参考: #

1.http://spark.apache.org/docs/latest/sql-performance-tuning.html