性能调优实战

本层定位

调优不是"背参数"。同一个 spark.sql.shuffle.partitions，在倾斜的表上是救命药，在均匀的表上是毒药。性能调优的本质是一套闭环方法论：先用可观测体系（监控、审计、Profile）找到瓶颈，再定位根因（是数据模型问题、配置问题还是资源问题），然后针对性地改配置或改模型，最后用回归验证确认修复生效并防止劣化回头。

• 找瓶颈（可观测）：没有度量就没有调优。吞吐、延迟、CPU/IO/内存水位、Shuffle 量、Spill 量必须先看得见（参见可观测页）；
• 定根因：区分三类根因——数据模型（分区键选错、倾斜键、宽表过宽）、配置（并行度、内存、文件大小）、资源（队列抢占、磁盘瓶颈）；
• 实施修复：一次只改一个变量，否则验证时说不清哪个改动起效；
• 回归验证：调优前后用同一数据集、同一并发做基准对比，指标（耗时/扫描量/费用）留档进调优知识库。

小文件治理

小文件是湖仓性能的头号慢性病。每个 Iceberg 数据文件在元数据中都有 manifest 条目、在引擎侧都要占用一次打开/关闭与调度开销。当单表文件数达到十万级，查询连"列清单"都要几秒钟，写入端 commit 变慢，NameNode（HDFS 部署时）内存被元数据压垮。

成因分析

• 高频提交：流式作业为了秒级可见性每分钟甚至每 10 秒 commit 一次，每次 commit 产生一批新文件，一天轻松堆出上万个文件；
• 分区过细：按"小时 × 城市 × 渠道"三重分区，低流量组合每天只落几 KB 数据，文件永远长不大；
• 并行写入过宽：Spark 写入 2000 个 task，每个 task 不管数据多少都各写一个文件，"众多个 1MB 文件"由此诞生；
• 上游切分：同步工具按块导出 CSV→Parquet，块大小即文件大小，未对齐目标文件尺寸。

检测：先量化再动手

-- ========== 检测 1：文件大小分布画像（基于 Iceberg 元数据表）==========
SELECT
  count(*)                                       AS total_files,
  round(sum(file_size_in_bytes) / 1024 / 1024)    AS total_mb,
  round(avg(file_size_in_bytes) / 1024 / 1024, 2) AS avg_mb,
  max(file_size_in_bytes) / 1024 / 1024           AS max_mb,
  sum(CASE WHEN file_size_in_bytes < 1048576     -- < 1MB 的小文件
           THEN 1 ELSE 0 END)                    AS files_lt_1m,
  round(sum(CASE WHEN file_size_in_bytes < 1048576
                 THEN file_size_in_bytes ELSE 0 END) * 100.0
        / sum(file_size_in_bytes), 1)            AS small_file_pct   -- 小文件字节占比
FROM lake.dwd_trade_order_di.files
WHERE partition LIKE '%dt=2026-11-04%';
-- 健康基线：avg_mb 接近目标文件大小(128-512MB)；files_lt_1m 占比 < 5%

-- ========== 检测 2：每个分区的文件数排行（找重灾区）==========
SELECT partition, count(*) AS file_cnt
FROM lake.dwd_trade_order_di.files
GROUP BY partition
ORDER BY file_cnt DESC
LIMIT 20;
-- 单分区文件数 > 1000 即列入合并候选

治标：rewrite_data_files 存储过程

Iceberg 自带的 rewrite_data_files 存储过程按策略把小文件重写为大文件，适合作为存量救火与例行合并手段：

-- binpack 策略（默认）：把小文件按目标大小装箱合并，开销最小
CALL lake.system.rewrite_data_files(
  table => 'lake.dwd_trade_order_di',
  strategy => 'binpack',
  where => 'dt = \'2026-11-04\'',                     -- 只合并当天分区，控制影响面
  options => map(
    'target-file-size-bytes', '268435456',              -- 目标 256MB/文件
    'min-file-size-bytes',    '10485760',               -- < 10MB 的才参与合并
    'max-file-size-bytes',    '536870912',              -- > 512MB 的大文件将被拆分
    'min-input-files',       '5'                        -- 一个分组至少 5 个文件才值得合并
  )
);

-- sort 策略：合并的同时按排序列重写，兼顾 Z-Order 聚簇（代价更高）
CALL lake.system.rewrite_data_files(
  table => 'lake.dwd_trade_order_di',
  strategy => 'sort',
  sort_order => 'user_id ASC NULLS FIRST',
  options => map('target-file-size-bytes', '268435456')
);

治本：写入端控制

合并是"吃药"，写入端控制才是"改生活方式"。三个抓手：目标文件大小、提交频率对齐、分区设计收敛。

# Flink 流式入湖：控制文件大小与提交频率（iceberg connector 参数）
table.exec.sink.upsert-materialize: AUTO
write.target-file-size-bytes: 268435456     # 256MB：文件写满即滚动
write.distribution-mode: hash              # 按 partition key hash 分区写入，避免每个 task 都写所有分区
commit.interval: 120000                    # 2 分钟一次 checkpoint/commit（原 10s 是万恶之源）
write.merge.schema: true

# Spark 批写入：先 repartition 落盘，再 commit，避免 2000 个小文件
# df.repartition(n, col("dt")) 的 n 按 日增量/256MB 估算

• target-file-size-bytes：设为 128–512MB。流式作业同时配 write.distribution-mode=hash，让同一分区的数据落进同一 writer，文件数 = 分区数 × 少数滚动，而不是 task 数 × 分区数；
• checkpoint 对齐：commit 间隔与可见性需求对齐——分钟级足够就别用秒级；真需要秒级可见，用"流表秒级 + 例行合并兜底"的组合拳；
• 分区设计原则：单分区日增量低于 1GB 就不要按天分区，改按月；组合分区维度优先用 Iceberg 隐藏分区（days(ts)、bucket(user_id, 16)）代替显式列分区。

Compaction 调度

合并不能靠人想起来才跑，要变成有触发条件、有资源预算、有验收指标的常态化调度。本节给出触发策略、DolphinScheduler 调度示例，以及与之配套的快照过期与孤儿文件清理——这三件事是"铁三角"，只做合并不做清理，元数据照样膨胀。

触发策略

调度示例：凌晨窗口 + Spark 存储过程

# DolphinScheduler：每日 01:30 合并窗口（含到期快照清理与孤儿文件清理）
name: iceberg_maintenance
cron: "0 30 1 * * ?"
tasks:
  - name: rewrite_small_files
    type: SPARK
    sparkParams: >
      --master yarn --deploy-mode cluster
      --queue maintenance           # 独立维护队列，不与业务抢资源
      --num-executors 10 --executor-cores 4 --executor-memory 8g
    script: |
      spark-sql -e "
        CALL lake.system.rewrite_data_files(
          table => 'lake.dwd_trade_order_di',
          strategy => 'binpack',
          options => map('target-file-size-bytes','268435456')
        )"

  - name: expire_snapshots                # 快照过期：物理删除被旧快照引用的文件
    type: SPARK
    depTasks: [rewrite_small_files]
    script: |
      spark-sql -e "
        CALL lake.system.expire_snapshots(
          table => 'lake.dwd_trade_order_di',
          older_than => timestamp '2026-10-29 00:00:00',
          max_concurrent_deletes => 4   # 限制并发删除，避免对象存储限流
        )"

  - name: remove_orphan_files             # 孤儿文件：无任何 manifest 引用的残留文件
    type: SPARK
    depTasks: [expire_snapshots]
    script: |
      spark-sql -e "
        CALL lake.system.remove_orphan_files(
          table => 'lake.dwd_trade_order_di',
          older_than => timestamp '2026-11-04 01:00:00'  # 仅清理 24h 前的孤儿，防止误删在途提交
        )"
# 顺序有讲究：先合并（产生新快照）→ 再过期（清旧快照）→ 最后扫孤儿（清无主文件）
# remove_orphan_files 的 older_than 必须留足缓冲，至少大于最大事务时长

查询加速

交互式查询的加速手段可以归纳为"五板斧"——缓存、分区裁剪、谓词下推、物化视图、排序聚簇。它们作用的环节不同，可以叠加使用，理解"每板斧省的是哪一段开销"是选型关键：

Iceberg 排序写入：提升裁剪率

数据物理有序后，每个文件在排序列上的 min/max 区间变窄，谓词下推能跳过更多文件。对高频过滤列做写入排序，是"改一次物理布局、所有查询受益"的投资：

-- 写入时声明排序：Spark SQL（对后续 INSERT 生效）
ALTER TABLE lake.dwd_trade_order_di
WRITE ORDERED BY (dt, user_id);
-- 排序列选择：优先高频过滤列（查询审计统计 Top 过滤列），
-- 其次低基数列在前、高基数在后，让文件区间收敛更快

-- 验证裁剪效果：对比排序前后扫描的文件数
SELECT count(*) AS files_scanned
FROM lake.dwd_trade_order_di.files
WHERE dt = '2026-11-04'
  AND 42 BETWEEN 1337 AND 1428;  -- 模拟 min/max 统计裁剪判定
-- 典型效果：按 user_id 查询单用户订单，文件命中率从 100% 降到 3–8%，
-- P99 查询延迟从 12s 降至 1s 以内

Doris 物化视图 + Trino 结果缓存

-- ========== Doris：异步物化视图（同步刷新型亦可）==========
CREATE MATERIALIZED VIEW lake.mv_user_gmv_1d
BUILD IMMEDIATE REFRESH SCHEDULE EVERY 1 HOUR
KEYS (user_id, dt)
DISTRIBUTED BY HASH(user_id) BUCKETS 10
PROPERTIES ("grace_period" = "0")
AS
SELECT user_id, dt,
       count(*)        AS order_cnt,
       sum(pay_amount) AS gmv
FROM lake.dwd_trade_order_di
WHERE pay_status = 1
GROUP BY user_id, dt;
# 查询命中 MV 时自动改写（query rewrite），对上层 SQL 完全透明

-- ========== Trino：结果缓存 + 元数据缓存（config.properties）==========
# query.max-history / result cache 属于企业发行版特性，
# 社区版可用 Coordinator 侧 catalog 缓存 + Alluxio 方案近似

# Trino config.properties：加速相关核心项
query.max-memory-per-node=8GB
# 驱动缓存：Iceberg 元数据缓存（减少 list/parse metadata 开销）
iceberg.catalog.caching.enabled=true
metadata.cache-ttl=1h
metadata.cache-maximum-size=10000
# 结果缓存（企业版 / 或前置一层数据网关实现同样语义）：
# 相同 SQL + 数据版本未变 → 直接返回缓存结果，Dashboard 场景命中率极高

资源隔离

性能问题的另一半根源不在"慢"，而在"抢"。一个分析师的全表扫描把队列占满，核心 ETL 全体延迟；报表高峰把 Trino worker 内存打爆，所有人的查询一起失败。资源隔离的目标是让重要负载不受不重要负载拖累，各归各的"车道"。

YARN 队列隔离

队列划分的要点（详细的公平/容量调度器对比见调度资源页）：按负载重要性分层，核心 ETL、临时查询、例行维护各占独立队列，配额与优先级区分：

<!-- capacity-scheduler.xml：三车道示例 -->
<property>
  <name>yarn.scheduler.capacity.root.queues</name>
  <value>etl,adhoc,maintenance</value>   <!-- 核心 / 临时 / 维护三车道 -->
</property>
<property>
  <name>yarn.scheduler.capacity.root.etl.capacity</name>
  <value>60</value>   <!-- ETL 稳态 60% -->
</property>
<property>
  <name>yarn.scheduler.capacity.root.etl.maximum-capacity</name>
  <value>80</value>   <!-- 峰值可弹性借用到 80% -->
</property>
<property>
  <name>yarn.scheduler.capacity.root.adhoc.capacity</name>
  <value>20</value>
</property>
<property>
  <name>yarn.scheduler.capacity.root.adhoc.maximum-capacity</name>
  <value>30</value>   <!-- adhoc 硬顶 30%，防止拖垮全局 -->
</property>
<property>
  <name>yarn.scheduler.capacity.root.maintenance.capacity</name>
  <value>20</value>
</property>

Trino resource groups

Trino 侧用 resource groups 管理并发与内存。规则按 user/source/catalog 匹配进组，每组限并发数与内存占比，超限排队：

# resource-groups.properties：选择器规则（admin 直通，其余入对应组）
resource-groups.configuration-manager=file
resource-groups.config-file=etc/resource_groups.json
# selector: user=etl_bot → group etl；user 匹配 ^ analyst_ → group adhoc

{
  "rootGroups": [
    {
      "name": "global",
      "softMemoryLimit": "80%",
      "softConcurrencyLimit": 40,
      "hardConcurrencyLimit": 60,
      "subGroups": [
        {
          "name": "etl",
          "softMemoryLimit": "50%",
          "softConcurrencyLimit": 10,      // 核心管道组
          "hardConcurrencyLimit": 15,
          "schedulingPolicy": "weighted_fair",
          "schedulingWeight": 100          // 高权重：调度优先保障
        },
        {
          "name": "adhoc",
          "softMemoryLimit": "20%",
          "softConcurrencyLimit": 20,      // 分析师交互查询组
          "hardConcurrencyLimit": 30,
          "schedulingPolicy": "fair",
          "schedulingWeight": 1            // 低权重：有空位才跑
        }
      ]
    }
  ],
  "selectors": [
    { "source": "etl-scheduler", "group": "global.etl" },
    { "user": "analyst_.*",      "group": "global.adhoc" }
  ]
}

Kafka / Spark 消费组隔离

• Kafka 消费组：同一 topic 的不同消费场景用不同 group.id——Flink 入湖、实时大屏、风控消费各自独立 offset，一个消费组积压不影响另外两个（但注意 group 变多会增加 broker 的元数据与 rebalance 开销）；
• Spark 消费隔离：spark streaming / 结构化流作业各自独占 executor 池（spark.yarn.queue + spark.dynamicAllocation 上限），禁止流批同队列混跑——批任务把流任务的 executor 挤掉，会造成 rebalance 风暴与 lag 堆积。

峰谷错峰策略

隔离解决"谁挤谁"，错峰解决"什么时候挤"。典型日内负载有两个峰：早 8–10 点报表峰、晚 20–22 点实时大屏峰。错峰手段：全量回刷/重算/归档等可延迟负载强制放 01:00–05:00 谷段（调度系统依赖编排保证先后）；adhoc 查询在大促等封网期临时降并发上限（resource group 热更新）；跨区报表错时拉数（华东 8 点跑、华北 9 点跑），把网络与查询峰摊开。

瓶颈排查方法论

面对"作业变慢了"这种模糊投诉，需要一套从粗到细的标准动线，避免"上来就翻日志"的低效排查。推荐四步法：

1. 看监控定位层：先看链路大盘——Kafka 消费 lag、Flink 算子反压、Spark stage 耗时、Trino 查询 P99。哪段曲线异常就往哪段钻，别在正常的层里翻日志；
2. 定位作业阶段：Spark UI（或 Spark History Server）看 Stage 累计时间与 GC/spill 指标；Trino 用 EXPLAIN ANALYZE 看算子级耗时分布；
3. 定位算子/倾斜：找"最长尾 Task"——如果 max task 时间是 median 的 10 倍以上，基本是数据倾斜；看该 task 的 shuffle read 大小确认；
4. 验证修复：按上文本页方法论闭环，改动与指标进调优知识库。

常见病症速查表

容量规划

容量规划是把"业务增量"翻译成"资源采购单"的算术题，也是防止"跑半年后存储与算力双不够"的前置防御。标准五步：数据增量测算 → 存储与计算反推 → 网络与峰值校验 → 压测验证 → 预留 30%。

1. 数据增量测算：按源系统口径统计"日增原始字节 × 副本与放大系数"。放大系数含：表格式与压缩（Parquet+ZSTD 通常 3–5× 压缩，即原始 1TB 落湖约 200–350GB）、多套副本（快照保留期内的增量、备份 1 份）、中间产物（Shuffle/临时目录，经验系数 1.3×）；
2. 存储与计算反推：存储按"N 个月在线 + 冷归档"折算总量；计算按"日批窗口需处理的总量 ÷ 可用窗口时间"反推核时；
3. 网络与峰值校验：核对跨机房同步带宽、对象存储吞吐上限（QPS 与 GB/s）、大促日峰值是否打穿链路最小环节；
4. 压测验证：用生产 3 倍日增的模拟数据走完整链路，验证 SLA 余量，而不是纸面推演；
5. 预留 30%：所有规划值 × 1.3，覆盖流量波动、故障重跑（大促后回刷）与突发的 ad-hoc 分析。

示例计算：日增 2TB 场景

-- ===== 输入假设 =====
原始日增量        = 2 TB（业务库/日志源合计）
Parquet 落湖压缩  ≈ 4:1（ZSTD）
副本/快照膨胀     = ×1.2（7 天快照保留）
备份副本          = ×1（跨区备份一份）
中间产物系数      = ×1.3（Shuffle/临时/重试）

-- ===== 存储推算（在线热/温层）=====
落湖日增量 = 2 TB / 4 = 0.5 TB
含快照膨胀 = 0.5 × 1.2 = 0.6 TB/天
在线保留 1 年（DWD 为主）≈ 0.6 × 365 ≈ 219 TB ≈ 220 TB
归档（冷层，另算）≈ 同量级但单价约 1/8

-- ===== 计算推算（日批窗口）=====
日处理量（含中间产物）= 0.5 TB × 1.3 ≈ 0.65 TB 实际扫描量更大（列存只扫必要列）
假设全链路日批需在 4h 窗口内完成，历史吞吐经验值 20 MB/s/核（含 IO 等待）：
所需核数 ≈ 0.65 TB × 1024 / 4h / (20 MB/s) ≈ 86 核持续
预留 30% ⇒ 约 120 核（再按 20% 失败重跑余量向上取整到机型规格）

-- ===== 网络校验 =====
跨区备份：0.5 TB/天，需 ≥ 10 MB/s 持续带宽（错峰 8h 窗口内完成）
对象存储读吞吐：大促峰值 3× 日增量集中回刷 ⇒ 需与云厂商确认 bucket 吞吐配额

调优案例

最后用一个真实形态的实战案例把全页知识串起来。背景实时入湖链路Flink → Iceberg

现象：端到端延迟告警

某电商平台"订单 CDC → Flink → Iceberg DWD → Trino 实时大屏"链路，稳定运行三个月后连续一周在晚高峰触发端到端延迟告警：Kafka 消费 lag 从常态 2 万涨到 80 万，大屏数据新鲜度从 1 分钟劣化到 15 分钟以上，且 Trino 侧大屏查询 P99 从 3s 涨到 20s。

定位：四步法走一遍

1. 看监控定位层：链路大盘显示 Kafka lag 堆积在 Flink 消费组，且 Flink 作业出现周期性反压——问题在消费侧处理能力，不在 broker；
2. 定位作业阶段：Flink WebUI 反压点在 Iceberg Sink 算子，该算子每次 checkpoint（10 秒一次）耗时从 2s 涨到 60s+，逼近 checkpoint 超时阈值；
3. 定位根因：查 Iceberg 表元数据发现单日分区文件数达 12000+（每个 subtask × 每 10 秒 commit 各产文件），manifest 膨胀导致 commit 变慢；同时 sink 算子按 order_id hash 后仍有 2 个大商家热点 key 造成写入倾斜；
4. 结论：根因是高频小文件 commit 叠加写入倾斜，二者互相放大——commit 慢导致 checkpoint 拖长，反压传导回消费端。

措施：三管齐下

# 措施 1：写入端治理（治本）——降频 + 定尺寸 + 按分区聚拢
commit.interval: 600000                      # 10s → 10min（大屏可接受分钟级新鲜度）
write.target-file-size-bytes: 268435456      # 256MB 目标文件
write.distribution-mode: hash                # hash 到分区键，聚拢写入

# 措施 2：倾斜治理——热点 key 加盐打散（Flink SQL 侧）
# sink 前按 (seller_id % 16) 预分区，把大商家流量摊到 16 个 subtask

# 措施 3：存量治理 + 例行维护（治标兜底）
# 每日 01:30 DolphinScheduler 调度：
#   rewrite_data_files(binpack) → expire_snapshots(7d) → remove_orphan_files

效果：前后指标对比