大数据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实现了可拆分性。