大数据SQL调优专题——底层调优

引入

上一篇我们提到了调优的常见切入点，核心就是通过数据产出情况发现问题，借助监控等手段收集信息排查瓶颈在哪，最后联合业务明确，等价重写思路去解决问题。
在实际工作场景中，去包管数据链路产出SLA的时候，我们都会先默认任务的SQL实现都是没问题的，也就是优先考虑，在不改写SQL语句的前提下，通过调解参数配置、优化数据存储等方法，提拔任务服从，减少实行时间。
针对这一类调优，我们通常称其为底层调优，下面我们就从参数和存储两方面去掌握。
参数调优

参数调优最简单的思路就是加资源，比如加内存，提高并行度等，来让任务更快的实行完成。除此之外，针对一些任务，我们可以通过调解框架自带的一些特定参数，能更有效的使作业运行服从达到最优。
这里只例举一下最常用的参数优化场景，针对差别数据处理引擎，差别业务场景，会有差别的参数优化方案，这一点在后续对应技术栈的专栏再去深入拓展。
并行实行

默认情况下，分布式任务会根据上下游依赖关系依次实行，每次仅实行一个任务。然而在某些条件下，如果任务之间没有直接的依赖关系，就意味着这些任务可以并行实行。此时通过合理配置并行实行的相干参数，就可以显著提高数据处理的服从和性能。具体参数的设置须要根据集群的硬件资源和数据量举行调解，才气达到最佳的并行处理结果。
Hive

相干参数：

• hive.exec.parallel：用于开启或关闭并行实行，默认值为 true。设置为 true 时，启用并行实行；设置为 false 时，禁用并行实行。
  
• hive.exec.parallel.thread.number：用于指定并行实行的线程数，默认值为 8。该参数决定了同时实行的任务数，可以根据集群的硬件资源情况举行调解。
  
• hive.exec.parallel.thread.queue.size：用于指定并行实行的线程队列大小，默认值为 0。当并行实行线程数达到上限时，新的任务会被放入队列中期待实行。
  

实现原理：

• Hive 的并行实行是通过将查询任务分解为多个子任务（Map 和 Reduce 任务），并在集群中的多个节点上并行实行这些子任务来实现的。每个 Map 任务处理一个数据片段，多个 Map 任务可以并行实行。Reduce 阶段也可以并行处理，差别的 Reduce 任务负责差别的数据分区。
  

优缺点：

• 优点：充实使用集群资源，加快查询速率，尤其实用于处理大规模数据集。
  
• 缺点：如果并行度过高，可能会导致资源竞争，如 CPU、内存等，从而影响整体性能；同时，任务之间的依赖关系可能会变得复杂，增加调试和维护的难度。
  

Spark

相干参数：

• spark.default.parallelism：它界说了没有显式指定并行度时的默认并行度。这个参数对于一些操纵如 RDD 的转换操纵、reduceByKey 等很重要。如果设置不妥，可能会导致资源使用不充实大概任务实行时间过长。通常可以根据集群的资源情况和数据规模举行调解。
  
• spark.sql.shuffle.partitions：这个参数决定了在 shuffle 操纵期间天生的分区数量。默认值通常取决于集群配置，但可以根据数据大小和集群资源举行调解。如果设置得太小，可能会导致数据倾斜，某些任务处理的数据量过大，从而延伸作业实行时间；如果设置得太大，会产生过多的小任务，增加任务调理开销。比方，在处理大规模数据集时，可以根据数据量和集群的核心数量举行合理调解，一般可以设置为集群核心数的 2-3 倍。
  
• spark.sql.adaptive.enabled：该参数用于启用自适应查询实行（AQE），允许 Spark 在运行时动态调解并行度。默认值为 false。
  
• spark.sql.adaptive.shuffle.targetPostShuffleInputSize：该参数用于设置 AQE 中每个 Shuffle 分区的目标大小（以字节为单位）。
  

实现原理：

• 并行度的设置：Spark SQL 通过设置 spark.sql.shuffle.partitions 和 spark.default.parallelism 参数来控制并行度。这些参数决定了数据在 Shuffle 操纵和每个 Stage 中的分区数，从而影响任务的并行实行。当数据量较大时，增加并行度可以提高查询性能，但过高的并行度可能会导致内存和网络资源的瓶颈。
  
• 自适应查询实行（AQE）：AQE 通过在运行时动态调解并行度来优化查询性能。当启用 AQE 时，Spark 会根据实际数据分布动态调解分区数，从而避免因分区数过小或过大导致的性能问题。AQE 会根据 spark.sql.adaptive.shuffle.targetPostShuffleInputSize 参数来决定是否归并分区，从而优化查询性能。
  

优缺点：

• 优点：
  
  
  
  • 提高查询性能：通过合理设置并行度，可以充实使用集群资源，提高查询速率。
    
  • 动态调解：启用 AQE 后，Spark 可以根据实际数据分布动态调解并行度，从而优化查询性能。
    
  • 灵活性：可以根据差别的查询需求和数据量，灵活调解并行度参数，以达到最佳性能。
    
  
  
• 缺点：
  
  
  
  • 资源消耗：过高的并行度可能会导致内存和网络资源的瓶颈，从而影响查询性能。
    
  • 动态调解的不确定性：启用 AQE 后，动态调解并行度可能会引入肯定的不确定性，须要仔细监控和调优，否则可能会出现反结果。
    
  
  

Flink

相干参数：

• parallelism.default：用于设置作业的并行度，可以在代码中通过设置实行环境的并行度来指定整个作业的并行实行水平。这个参数可以控制任务实行的并发度，根据硬件资源和数据量的大小举行调解。
  
• table.exec.resource.default-parallelism：针对Flink SQL任务设置默认并行度，需与集群资源匹配，避免过高导致资源争抢。
  

实现原理：

• 数据分区：Flink 将输入数据举行分区，以便多个任务可以并行处理差别的数据分区。常见的分区方式有哈希分区、范围分区等。比方，在举行聚合操纵时，Flink 可以根据指定的键将数据举行哈希分区，使得雷同键的数据被分配到同一个任务中举行处理。
  
• 任务调理：Flink 的任务调理器负责将任务分配到差别的任务槽（Task Slot）中实行。任务槽是 Flink 中用于实行任务的资源单位，可以明确为一个实行任务的容器。任务调理器会根据任务的并行度和可用的任务槽资源，将任务分配到差别的节点上实行。
  
• 数据互换：在并行实行过程中，差别任务之间须要举行数据互换。Flink 提供了多种数据互换的方式，如广播（Broadcast）、重新分区（Repartitioning）等。比方，在举行连接操纵时，Flink 可能须要将两个输入数据集举行重新分区，以便举行连接操纵。
  

优缺点：

• 优点：
  
  
  
  • 提高处理性能：通过并行实行，可以充实使用多核处理器和分布式盘算环境，提高数据处理的速率和吞吐量。多个任务可以同时处理差别的数据分区，从而减少处理时间。
    
  • 可扩展性强：Flink 的并行实行机制使得它可以很容易地扩展到大规模数据集和分布式盘算环境中。可以根据数据量和盘算资源的变革，动态调解并行度，以满足差别的处理需求。
    
  • 容错性好：Flink 具有良好的容错机制，当某个任务失败时，它可以自动重新启动失败的任务，并从查抄点（Checkpoint）恢复状态。在并行实行环境中，纵然某个任务失败，其他任务仍然可以继续实行，不会影响整个作业的运行。
    
  
  
• 缺点：
  
  
  
  • 资源竞争：在并行实行过程中，多个任务可能会竞争盘算资源，如 CPU、内存等。如果资源分配不合理，可能会导致某些任务实行迟钝，影响整体性能。
    
  • 数据倾斜：数据倾斜是指数据在差别任务之间分布不匀称的情况。如果某些任务处理的数据量远远大于其他任务，可能会导致这些任务实行时间过长，影响整个作业的性能。
    
  
  

预聚合

在一些聚合处理场景中，数据读取和扫描之后，引擎可以先举行一次预聚合，然后再将聚合结果举行分发和Shuffle。这样做可以减少在数据分发过程中的时间消耗和资源开销，从而加快查询历程。通过合理配置相干参数，可以显著提高数据处理的服从和性能。
Hive

相干参数：

• hive.map.aggr：控制是否启用 Map 端预聚合。当设置为 true 时，Hive 会在 Map 阶段对数据举行局部聚合，从而减少传输到 Reduce 阶段的数据量，提高查询性能。
  
• hive.groupby.mapaggr.checkinterval：设置在 Map 端预聚合时，查抄聚合键数量的间隔。当 Map 端处理的数据行数达到这个值时，Hive 会查抄聚合键的数量。如果聚合键数量高出肯定比例，Hive 会认为预聚合结果不抱负，从而关闭预聚合功能。
  
• hive.groupby.skewindata：控制是否启用数据倾斜处理。当设置为 true 时，Hive 会在处理数据倾斜时优化 Group By 操纵，从而提高查询性能。
  

实现原理：

• Hive 的预聚合是在 Map 阶段对数据举行局部聚合，减少数据传输量和盘算量。在 Map 阶段，对每个分组键举行局部聚合，天生中间结果，然后将这些中间结果通报给 Reduce 阶段举行最终聚合。
  

优缺点：

• 优点：显著提高查询服从，减少数据传输和盘算开销。
  
• 缺点：一些如盘算中位数等的场景中，会导致最闭幕果不正确；增加内存使用量，内存不足时可能影响作业实行。
  

Spark

相干参数：

• spark.sql.autoBroadcastJoinThreshold：这个参数用于控制自动广播连接的阈值。当一个表的数据量小于这个阈值时，Spark 会自动将该表广播到全部的实行节点上，以避免 Shuffle 操纵，从而提高连接操纵的性能。在预聚合场景中，如果小表可以被广播，那么可以在连接之前对小表举行预聚合，进一步提高性能。
  
• spark.sql.shuffle.partitions：该参数决定了 Shuffle 操纵时的分区数。合理设置分区数可以避免数据倾斜，提高预聚合的服从。如果分区数过多，会导致过多的小文件，增加管理开销；如果分区数过少，可能会导致数据倾斜，影响性能。
  
• spark.sql.inMemoryColumnarStorage.batchSize：这个参数控制内存列存储的批处理大小。在预聚合过程中，合理调解这个参数可以优化内存使用和性能。较大的批处理大小可以减少内存开销，但可能会增加内存压力；较小的批处理大小可以更好地适应内存限定，但可能会降低性能。
  

实现原理：

• 广播连接：当连接的表大小小于 spark.sql.autoBroadcastJoinThreshold 阈值时，Spark 会自动将小表广播到全部节点，从而减少数据传输和 Shuffle 操纵。通过广播小表，可以避免数据的 Shuffle 操纵，从而提高连接操纵的性能。广播连接实用于小表与大表的连接场景。
  
• Shuffle 分区数：通过设置 spark.sql.shuffle.partitions 参数，可以控制 Shuffle 操纵的分区数，从而影响数据的并行处理本领。增加 Shuffle 分区数可以提高并行度，从而加快查询速率，但过高的并行度可能会导致内存和网络资源的瓶颈。
  

优缺点：

• 优点：
  
  
  
  • 提高查询性能：通过提前对数据举行聚合操纵，可以减少后续盘算的工作量，从而提高查询性能。特别是在处理大规模数据和复杂查询时，预聚合可以显著缩短查询实行时间。
    
  • 减少数据传输和存储开销：预聚合可以减少中间结果的数据量，从而减少数据传输和存储开销。这对于分布式盘算环境非常重要，因为数据传输和存储开销往往是性能瓶颈之一。
    
  • 优化内存使用：通过合理配置预聚合相干参数，可以优化内存使用，避免内存溢出等问题。比方，可以调解 spark.sql.inMemoryColumnarStorage.batchSize 参数来控制内存列存储的批处理大小，以适应差别的内存限定。
    
  
  
• 缺点：
  
  
  
  • 增加盘算开销：预聚合须要额外的盘算资源，特别是在数据量较大时，预聚合的盘算开销可能会比较大。如果预聚合的收益不足以抵消盘算开销，那么可能会导致性能下降。
    
  • 可能导致数据倾斜：如果预聚合的结果不匀称，可能会导致数据倾斜。比方，如果某些分组的聚合结果非常大，而其他分组的聚合结果非常小，那么在后续的盘算中可能会出现数据倾斜，影响性能。
    
  • 不适合动态数据：预聚合通常实用于静态数据或数据变革不频仍的场景。如果数据变革频仍，那么预聚合的结果可能会很快失效，须要重新举行盘算，从而增加盘算开销。
    
  
  

Flink

在Flink聚合操纵可以通过启用微批处理和Local-Global计谋来优化。这种方法将原本会合的聚合操纵分解为两个阶段，首先在本地举行预聚合，然后在全局范围内再次举行聚合。这样做可以减少状态访问次数，提高处理吞吐量，并减少数据输出量。
微批处理（MiniBatch）
实现原理：
微批处理通过缓存肯定量的数据，然后一次性处理这些数据，从而减少对状态（State）的访问频率，提拔吞吐量并减少数据的输出量。这种方式通过增加少量的延迟来调换更高的吞吐量。
参数配置：

• table.exec.mini-batch.enabled：是否开启微批处理，默认值为 false。
  
• table.exec.mini-batch.allow-latency：批量输出数据的时间间隔，比方 5s。
  
• table.exec.mini-batch.size：微批操纵所缓存的最大数据条数，比方 20000。
  

优缺点：

• 优点：提拔吞吐量，减少对状态的访问，实用于一般聚合场景。
  
• 缺点：增加延迟，不实用于对延迟要求极高的场景。
  

Local-Global 计谋
原理：
Local-Global 计谋将聚合操纵分为两个阶段：Local 聚合和 Global 聚合。第一阶段在上游节点本地对数据举行聚合（LocalAgg），并输出这次微批的增量值（Accumulator）。第二阶段再将收到的 Accumulator 归并（Merge），得到最终的结果（GlobalAgg）。这种计谋类似于 MapReduce 模子中的 Combine + Reduce 处理模式，通过本地聚合筛除部分倾斜数据，从而降低 Global 聚合的热点，提拔性能。
参数配置：

• table.optimizer.agg-phase-strategy：聚合阶段的计谋，可选值为 AUTO、TWO_PHASE（使用 Local-Global 两阶段聚合）和 ONE_PHASE（仅使用 Global 一阶段聚合）。默认值为 AUTO。
  

优缺点：

• 优点：有效解决数据倾斜问题，提拔聚合性能，实用于普通聚合操纵。
  
• 缺点：须要聚合函数支持 Merge 方法，且在某些情况下可能结果不明显。
  

   在默认设置下，聚合算子会对每条流入的数据实行一系列操纵，比方读取状态、更新状态。当数据量较大时，这些状态操纵的开销也会相应增加，从而影响整体服从，这种影响在使用如RocksDB这类序列化本钱较高的State Backend时尤为明显。而一旦启用了Mini-Batch处理，流入的数据会被暂存于算子内部的缓冲区中，直到达到预设的容量或时间阈值，然后才会举行聚合逻辑处理。这种方法使得同一批数据中的每个唯一键(key)只需举行一次状态的读写操纵，特别是在键分布较为稀疏的情况下，这种优化的结果会更为显著。
  数据重用

如果查询语句或任务中存在重复的查询块（即中间结果集）或子查询，可以采取数据重用（Reuse）计谋。比方根据差别的过滤条件多次筛选雷同的表，每筛选一次就须要重新扫描一次表，通过数据缓存等方式，相较于之前多次扫描数据，则只需举行一次扫描，从而减少对表的读取次数。这样不仅降低了读写I/O和网络的开销，还能加快任务的实行速率，缩短产出时间。差别于常见的数据缓存（Cache），数据重用强调的是在程序中重复使用已盘算的对象或结果，而缓存则是具体实现重用的一种机制。
Hive

参数配置：

• hive.optimize.cte.materialize.threshold：当设置为大于 0 的值时，如果一个 CTE（公共表表达式） 被引用的次数高出该阈值，Hive 会将该 CTE 的结果物化（即存储到临时表中），以避免重复盘算。
  

   CTE就是使用WITH AS语句
  实现原理：

• 物化机制：当 CTE 的引用次数高出 hive.optimize.cte.materialize.threshold 的值时，Hive 会在实行主查询之前，将 CTE 的结果存储到一个临时表中。
  
• 实行流程：
  
  
  
  • Hive 在解析 SQL 时，会查抄 CTE 的引用次数。
    
  • 如果引用次数高出阈值，Hive 会天生一个临时表来存储 CTE 的结果。
    
  • 后续的查询可以直接从临时表中读取数据，而不须要重新盘算 CTE。
    
  
  

优缺点：

• 优点：
  
  
  
  • 提高查询服从：通过物化 CTE 的结果，减少了重复盘算的开销，特别是在 CTE 被多次引用的情况下，性能提拔显著。
    
  • 简化代码：使用 CTE 可以使 SQL 代码更加轻巧和易读，同时通过物化避免了性能丧失。
    
  
  
• 缺点：
  
  
  
  • 存储开销：物化会占用额外的存储空间，特别是当 CTE 的结果集较大时，可能会导致存储压力。
    
  • 实用性限定：仅在 CTE 被多次引用时才有显著结果，如果 CTE 仅被引用一次，物化可能不会带来性能提拔。
    
  
  

Spark

下令使用：
CACHE TABLE 是 Spark SQL 中的一个下令，用于将表的内容或查询结果缓存到内存或磁盘中，以提高后续查询的性能。通过缓存数据，可以减少对原始数据源的重复读取，从而加快查询速率。
实现原理：

• 缓存机制：CACHE TABLE 下令会将指定的表或查询结果存储在内存或磁盘中。Spark SQL 会根据配置的存储级别（如 MEMORY_ONLY、MEMORY_AND_DISK 等）来决定数据的存储方式。
  
• 存储级别：可以通过 OPTIONS 子句指定存储级别。比方，MEMORY_AND_DISK 表示数据优先存储在内存中，如果内存不足则存储在磁盘上。
  
• 查询优化：缓存的数据会在后续查询中直接使用，避免了重复的盘算和 I/O 操纵，从而提高查询性能。
  

优缺点：

• 优点：
  
  
  
  • 提高查询性能：通过缓存数据，减少了对原始数据源的重复读取，加快了查询速率，特别是在多次查询同一数据集的情况下，性能提拔显著。
    
  • 灵活的存储级别：可以根据实际需求选择差别的存储级别，以平衡性能和存储资源的使用。
    
  
  
• 缺点
  
  
  
  • 存储资源消耗：缓存数据会占用内存或磁盘资源，如果数据量较大，可能会导致存储压力，影响系统的整体性能。
    
  • 缓存一致性问题：如果底层数据发生变革，缓存的数据可能不再一致，须要手动刷新或清除缓存，以确保数据的正确性。
    
  
  

Flink

对Flink来说，基于其流式盘算的底层计划，数据重用计谋主要通过内存、Redis 和 RocksDB 等方式实现缓存。内存缓存通过内部内存去存储数据，实现数据重用；Redis 和RocksDB缓存则是通过外部键值存储实现数据共享和通信。
存储调优

存储调优的思路，主要是接纳合适的存储格式和压缩类型，来减少大数据处理中须要传输的文件大小，从而提高任务完成服从。
存储格式

常见存储格式如下：
文本格式（TextFile）：这是最基本的存储格式，数据以纯文本形式存储，每行一条记载。优点是简单直观，易于明确和检察，可直接使用文本编辑器打开。缺点是存储服从低，占用空间大，而且在读取和处理时相对较慢。比方，在处理大规模数据集时，读取文本格式的数据可能会消耗大量的时间和资源。
JSON 格式：JSON 是一种轻量级的数据互换格式，它在数据处理中也被广泛使用。在 Spark 中，可以使用 JSON 格式存储数据。JSON 格式具有良好的可读性和可扩展性，但是存储服从相对较低。在对数据的可读性要求较高大概须要与其他系统举行数据互换时，可以考虑使用 JSON 格式。
CSV （Comma-Separated Values）格式：CSV格式是一种常见的表格数据存储格式，它以逗号分隔各个字段。在 Spark 中，可以使用 CSV 格式存储数据。CSV 格式简单直观，易于明确和处理，但是存储服从较低，而且在处理大规模数据集时可能会遇到性能问题。
ORC（Optimized Row Columnar）格式：ORC 是一种高效的列式存储格式，它对数据举行了优化存储，能够提高数据的读取和写入服从。与文本格式相比，ORC 格式占用的存储空间更小，查询性能更好。ORC 文件支持复杂数据类型和嵌套结构，而且可以举行高效的压缩。在 Hive 中使用 ORC 格式可以显著提高查询性能，特别是对于大规模数据集。
Parquet 格式：Parquet 也是一种列式存储格式，它具有高效的压缩比和良好的查询性能。Parquet 格式支持多种压缩算法，如 Snappy、Gzip 和 Zstandard（zstd）等。通过选择合适的压缩算法，可以进一步减小存储占用空间并提高查询服从。在 Hive 中，使用 Parquet 格式联合适当的压缩算法可以提高存储服从和查询性能。
Avro 格式：Avro 是一种数据序列化系统，它具有良好的兼容性和可扩展性。在 Flink 中，可以使用 Avro 格式存储数据。Avro 格式支持复杂数据类型和嵌套结构，而且可以举行高效的压缩。通过使用 Avro 格式，可以方便地与其他系统举行数据互换和集成。
压缩类型

Gzip 压缩：最早由Jean-loup Gailly和Mark Adler创建，用于UNIX系统的文件压缩。我们在Linux中经常会用到后缀为.gz的文件，它们就是GZIP格式的。Gzip 是一种广泛使用的压缩算法，它具有较高的压缩比，可以有效地减小存储占用空间。但是，Gzip 压缩息争压缩的速率相对较慢，可能会对数据处理的性能产生肯定的影响。在对存储空间要求较高而对性能要求相对较低的场景下，可以考虑使用 Gzip 压缩。
Deflate 压缩：Deflate 是一种通用的压缩算法，它在压缩比和压缩息争压缩速率之间取得了较好的平衡。
Snappy 压缩：是一款由Google开发的开源压缩库。它的目的不是最大水平地压缩或与任何其他压缩库兼容，相反，它旨在实现高速且合理的压缩。比方，与Gzip的最快模式相比，Snappy的压缩速率对于大多数输入来说要快一个数量级，但天生的压缩文件要大20%～100%。
LZO（Lempel-Ziv-Oberhumer） 压缩：LZO是一种快速的压缩算法，它在压缩息争压缩速率方面与 Snappy 类似。它能够提供非常快速的压缩息争压功能。解压并不须要内存的支持，纵然使用非常大的压缩比例迟钝压缩出的数据，依然能够非常快速地解压。LZO遵循GPL。
LZ4压缩：LZ4是无损压缩算法，提供每核大于500MB/s的压缩速率，可通过多核CPU举行扩展。它具有极快的解码器，每个内核的速率可达数GB/s，通常在多核系统上达到RAM速率限定。一方面，LZ4的速率可以动态调解，即选择一个“加快”因子，以压缩比调换更快的速率。另一方面，它还提供了高压缩率的衍生产品LZ4_HC，以CPU时间调换更高的压缩率。全部版本都具有雷同的解压缩速率。
Zstandard（zstd）压缩：Zstandard是Facebook在2016年开源的一种快速、无损压缩算法，简称zstd算法，实用于实时压缩场景，并拥有更好的压缩比。相比业内其他压缩算法，zstd算法的特点是当须要时，它可以将压缩速率互换为更高的压缩比率（压缩速率与压缩比率的权衡可以通过小增量来配置）​。
   关于压缩可以看看之前我写的这篇文章。
  在大数据开源系统中，Hadoop、Kafka、Pulsar等都支持多少种压缩协议，ORC、Parquet、Avro等文件格式也支持此中大部分压缩算法。在大数据系统中，尤其是在HDFS中，压缩后的文件是否可拆分也是一个重要考量尺度，LZO、LZ4等格式是支持拆分的，Zstandard也通过Hadoop 4mc实现了可拆分性。

免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！更多信息从访问主页：qidao123.com:ToB企服之家，中国第一个企服评测及商务社交产业平台。