计算引擎层

本层定位

湖仓的计算层只回答一个问题：谁来算、以什么节奏算。因为存储层已经解耦（见 存储与表格式层），计算引擎可以按负载自由选择与替换——这正是“存算分离”红利的兑现处。 业界事实标准是“Spark + Flink”双引擎分工：

• Spark 面向大规模离线批处理：T+1 数仓回刷、全量重算、机器学习训练、图计算； 吞吐之王，作业即起即走。
• Flink 面向持续低延迟流处理：CDC 实时入湖、实时打宽、CEP 复杂事件、 秒级大屏指标；状态管理是一等公民，作业常驻运行。

“批流一体”的核心不在引擎本身，而在共享的表：Flink 把变更流以主键 upsert 方式持续写入 Iceberg， Spark 与 Trino 按 snapshot 读取同一张表的稳定视图——流与批读的永远是一份物理数据，不再需要“实时表 + 离线表”两套口径。

Apache Spark

架构总览

与 Iceberg 集成

Spark 是 Iceberg 生态最完整的引擎：SQL 扩展（MERGE INTO / 时间旅行 / 存储过程）开箱即用。 集成三要素：runtime 包、catalog 定义、SQL 扩展。

# spark-defaults.conf：Iceberg on Spark 三要素
spark.sql.extensions=org.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions
spark.sql.catalog.lakehouse=org.apache.iceberg.spark.SparkCatalog
spark.sql.catalog.lakehouse.type=hive
spark.sql.catalog.lakehouse.uri=thrift://hms.lakehouse.local:9083
spark.sql.catalog.lakehouse.warehouse=s3a://lakehouse/warehouse

建表、MERGE INTO 与时间旅行的完整 SQL 见存储层 · Apache Iceberg一节。 引擎侧最常用的维护能力是存储过程（Stored Procedures），compaction 是其中最核心的一个：

-- binpack（默认）：把小文件打包合并到 target 大小，代价最小
CALL lakehouse.system.rewrite_data_files(
  table    => 'lakehouse.dws_orders',
  strategy => 'binpack',
  where    => 'dt = DATE '\''2026-11-04'\'''     -- 只处理当天分区，避免全表重写
);

-- sort：合并的同时按排序列重写，提高文件级裁剪命中率
CALL lakehouse.system.rewrite_data_files(
  table      => 'lakehouse.dws_orders',
  strategy   => 'sort',
  sort_order => 'city ASC NULLS FIRST, order_ts DESC'
);

-- zorder：多维聚簇，适合多个“等权重”查询列的组合过滤
CALL lakehouse.system.rewrite_data_files(
  table      => 'lakehouse.dws_orders',
  strategy   => 'zorder',
  sort_order => 'zorder(city, user_id, order_ts)'
);

-- 配套维护：manifest 合并与快照过期（与存储层治理章节同一套动作）
CALL lakehouse.system.rewrite_manifests(table => 'lakehouse.dws_orders');
CALL lakehouse.system.expire_snapshots(
  table => 'lakehouse.dws_orders',
  older_than => TIMESTAMP '2026-10-28 00:00:00',
  retain_last => 100
);

spark-submit 提交模板

spark-submit --master yarn --deploy-mode cluster \
  --name dws_orders_merge_$(date +%F) \
  --packages org.apache.iceberg:iceberg-spark-runtime-3.5_2.12:1.5.2 \
  --conf spark.sql.extensions=org.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions \
  --conf spark.sql.catalog.lakehouse=org.apache.iceberg.spark.SparkCatalog \
  --conf spark.sql.catalog.lakehouse.type=hive \
  --conf spark.sql.catalog.lakehouse.uri=thrift://hms.lakehouse.local:9083 \
  --conf spark.sql.catalog.lakehouse.warehouse=s3a://lakehouse/warehouse \
  --conf spark.hadoop.fs.s3a.endpoint=http://minio.lakehouse.local:9000 \
  --conf spark.hadoop.fs.s3a.path.style.access=true \
  --conf spark.hadoop.fs.s3a.committer.name=magic \
  --conf spark.sql.adaptive.enabled=true \
  --conf spark.sql.shuffle.partitions=800 \
  --num-executors 20 \
  --executor-cores 4 \
  --executor-memory 16g \
  --conf spark.memory.fraction=0.80 \
  --conf spark.executor.memoryOverhead=4g \
  --class com.lakehouse.etl.DwsOrdersJob \
  /opt/jobs/lakehouse-etl-1.0.0.jar --date 2026-11-04

核心调优参数表

AQE：开启即得的性价比之王

# 自适应查询执行（Spark 3+）：基于运行时统计自动纠偏
spark.sql.adaptive.enabled=true
spark.sql.adaptive.coalescePartitions.enabled=true
spark.sql.adaptive.coalescePartitions.initialPartitionNum=1600   # 起始分区给大，让 AQE 往回合
spark.sql.adaptive.advisoryPartitionSizeInBytes=134217728        # 合并目标 128MB
spark.sql.adaptive.skewJoin.enabled=true                         # 倾斜 join 自动拆分热点分区
spark.sql.adaptive.skewJoin.skewedPartitionFactor=5
spark.sql.adaptive.skewJoin.skewedPartitionThresholdInBytes=268435456
spark.sql.adaptive.localShuffleReader.enabled=true              # 避免 shuffle 的本地读

数据倾斜处理

症状：个别 Task 执行几十分钟而其余早已完成（Spark UI 可见 Stage 内 Task 耗时呈长尾）。三套打法按序尝试：

-- ============================================================
-- 场景：dwd_orders（大表）JOIN dim_user（维表，热点 key 倾斜）
-- ============================================================

-- 方案 A：小表可放内存 → 直接广播，join 退化为 map 端操作（首选）
SELECT /*+ BROADCAST(u) */ o.order_id, o.user_id, u.city
FROM lakehouse.dwd_orders o
JOIN lakehouse.dim_user u ON o.user_id = u.user_id
WHERE o.dt = '2026-11-04';

-- 方案 B：salting 加盐 —— 把倾斜 key 打散成 N 个桶分摊计算
WITH big_salted AS (
  SELECT o.*, pmod(hash(o.user_id), 64) AS salt     -- 大表：key 加盐到 64 桶
  FROM lakehouse.dwd_orders o
  WHERE o.dt = '2026-11-04'
), small_replicated AS (
  SELECT u.*, s.salt                              -- 维表：按同样规则复制 64 份
  FROM lakehouse.dim_user u
  LATERAL VIEW explode(sequence(0, 63)) s AS salt
)
SELECT b.order_id, b.user_id, b.order_amount
FROM big_salted b
JOIN small_replicated r ON b.user_id = r.user_id AND b.salt = r.salt;

-- 方案 C：两阶段聚合 —— 先局部聚合压缩数据量，再全局聚合
SELECT city, SUM(pv) AS pv, SUM(gmv) AS gmv
FROM (
  SELECT user_id, city,
         count(*)     AS pv,          -- map 端局部聚合：热点 key 先被压缩
         sum(amount)  AS gmv
  FROM lakehouse.dwd_orders
  WHERE dt = '2026-11-04'
  GROUP BY user_id, city
)
GROUP BY city;

Apache Flink

架构总览

• JobManager：作业主控。JobMaster 负责调度与 CheckpointCoordinator 协调快照； ResourceManager 管理 TaskManager/Slot 资源；Dispatcher 接收提交、拉起 JobManager。
• TaskManager（TM）：工作进程，内部划分若干 slot（资源切片，同作业算子可共享）； 每个 Task 消费中间结果分区，承接网络传输与状态访问。
• Checkpoint：基于 Chandy-Lamport 分布式快照——Barrier 随数据流入算子， 对齐后把各算子状态异步持久化到远端（对象存储/HDFS），失败时全体回滚到最近一致点。

流表一体的核心：Checkpoint + 两阶段提交写 Iceberg

Flink 的 Checkpoint 只保证引擎内部状态一致；要把一致性延伸到 Iceberg 表， 靠的是 sink 的两阶段提交（2PC）协议：

1. 预提交（prepareCommit）：Writer 把本间隔数据写成 Parquet 数据文件与删除文件，登记为 pending；
2. 状态持久化：pending 清单进入 Checkpoint 状态后端——这一步成功，提交才“有据可查”；
3. 正式提交（commit）：收到 Checkpoint 完成通知后，Committer 生成新 snapshot 并原子切换 metadata.json 指针；这与 Iceberg 的提交协议（见存储层）完全同构；
4. 故障恢复：从 Checkpoint 重启后，既不会丢已确认数据（Exactly-Once），也不会提交半成品（孤儿文件另由维护作业清理）。

Watermark 与事件时间

乱序数据流上，“现在几点”由 Watermark 声明：watermark = max(event_ts) − 允许迟到。 窗口只有在 watermark 越过其结束边界后才触发计算——这是事件时间语义下结果正确性的基石：

-- Kafka 源表：事件时间戳 + 5 秒乱序容忍
CREATE TABLE kafka_orders (
  order_id   STRING,
  order_ts   TIMESTAMP(3),
  amount     DECIMAL(12, 2),
  WATERMARK FOR order_ts AS order_ts - INTERVAL '5' SECOND
) WITH (
  'connector' = 'kafka',
  'topic' = 'orders',
  'properties.bootstrap.servers' = 'kafka1:9092,kafka2:9092',
  'properties.group.id' = 'dws_gmv_job',
  'scan.startup.mode' = 'latest-offset',
  'format' = 'json'
);

-- 事件时间滚动窗口：窗口闭合才产出，迟到 5 秒内数据完整计入
SELECT
  DATE_FORMAT(window_start, 'yyyy-MM-dd HH:mm') AS win_start,
  count(*)    AS order_cnt,
  sum(amount) AS gmv
FROM TABLE(
  TUMBLE(TABLE kafka_orders, DESCRIPTOR(order_ts), INTERVAL '1' MINUTE)
)
GROUP BY window_start, window_end;

实战：Flink SQL CDC → Iceberg 完整作业

-- ============================================================
-- MySQL CDC 实时入湖：全量快照 + 增量 binlog，一作业搞定
-- ============================================================

-- 0) 作业级配置（SQL 客户端）
SET 'execution.checkpointing.interval' = '2min';
SET 'execution.checkpointing.mode'     = 'EXACTLY_ONCE';
SET 'pipeline.parallelism'             = '8';

-- 1) 注册 Iceberg Catalog（指向存储层部署的 HMS）
CREATE CATALOG iceberg_hms WITH (
  'type'         = 'iceberg',
  'catalog-type' = 'hive',
  'uri'          = 'thrift://hms.lakehouse.local:9083',
  'warehouse'    = 's3a://lakehouse/warehouse',
  's3.endpoint'  = 'http://minio.lakehouse.local:9000',
  's3.path.style.access' = 'true'
);

-- 2) CDC 源表：primary key 让引擎按 key 路由，更新走 retract/upsert 通道
CREATE TABLE orders_src (
  order_id     BIGINT,
  user_id      STRING,
  order_amount DECIMAL(12, 2),
  order_ts     TIMESTAMP(3),
  op_type      STRING,
  PRIMARY KEY (order_id) NOT ENFORCED
) WITH (
  'connector'         = 'mysql-cdc',
  'hostname'          = 'mysql.lakehouse.local',
  'port'              = '3306',
  'username'          = 'flink_cdc',
  'password'          = '******',
  'database-name'     = 'shop',
  'table-name'        = 'orders',
  'scan.startup.mode' = 'initial',        -- 先全量快照，再无缝接 binlog
  'server-time-zone'  = 'Asia/Shanghai'
);

-- 3) 目标表：Iceberg 主键表（分区列必须包含在主键中，upsert 才能定位分区）
CREATE TABLE iceberg_hms.lakehouse.dws_orders (
  order_id     BIGINT,
  user_id      STRING,
  order_amount DECIMAL(12, 2),
  order_ts     TIMESTAMP(3),
  dt           DATE,
  PRIMARY KEY (order_id, dt) NOT ENFORCED
) PARTITIONED BY (dt)
WITH (
  'format-version'               = '2',
  'write.format.default'         = 'parquet',
  'write.target-file-size-bytes' = '134217728',   -- 128MB，流式写入适当小于批
  'write.upsert.enabled'         = 'true',        -- 主键 upsert：变更只改受影响记录
  'write.metadata.delete-after-commit.enabled' = 'true'
);

-- 4) 持续同步：每 2 分钟一次 Checkpoint 即一次 Iceberg 提交
INSERT INTO iceberg_hms.lakehouse.dws_orders
SELECT order_id, user_id, order_amount, order_ts, DATE(order_ts) AS dt
FROM orders_src;

Checkpoint 关键配置

# flink-conf.yaml：Checkpoint 与状态后端
execution.checkpointing.interval: 2 min          # 提交间隔 = 数据可见性延迟
execution.checkpointing.mode: EXACTLY_ONCE
execution.checkpointing.timeout: 10 min
execution.checkpointing.max-concurrent-checkpoints: 1
execution.checkpointing.min-pause: 30 s          # 两次快照间的最小间隔，防“追快照”
state.backend: rocksdb                           # 大状态首选（1.19+ 改名 state.backend.type）
state.backend.incremental: true                  # 增量快照，只上传变化 SST
state.checkpoints.dir: s3://flink/checkpoints
state.checkpoints.num-retained: 3

批流一体设计

Kappa vs Lambda

“流写批读”的落地模式

两股写入并发作用于同一张表时，Iceberg 的乐观并发机制会校验冲突：不同分区的写直接并行成功； 同分区的“流 upsert + 批 overwrite”则后者提交时校验失败并重试——把批回刷设计在独立分区（如按 dt 刷历史） 可以从根上避开冲突重试。

Mini-Batch 写入：提交间隔与 Checkpoint 对齐

Flink 的 Iceberg sink 只在 Checkpoint 时提交，因此“提交间隔 = Checkpoint 间隔”，三者必须联动规划：

• 间隔过短（如 30s）：每分钟 2 个快照，manifest/小文件爆炸，查询 planning 陡增；
• 间隔过长（如 30min）：下游实时性不可接受，故障时状态回放代价大；
• 经验公式：间隔内单分区写入量 ÷ target-file-size ≈ 1~2 个文件， 反推合理间隔（生产常见 1–5 分钟起步）。

-- upsert 与可见性的关键开关（表属性或 OPTIONS 提示二选一）
INSERT INTO iceberg_hms.lakehouse.dws_orders
/*+ OPTIONS(
      'write.upsert.enabled' = 'true',            -- 主键表记录级更新，不重写整个文件
      'write.target-file-size-bytes' = '134217728'
   ) */
SELECT order_id, user_id, order_amount, order_ts, DATE(order_ts) AS dt
FROM orders_src;

-- 作业侧：提交节奏与 Checkpoint 严格对齐，二者是同一个东西
SET 'execution.checkpointing.interval' = '2min';
SET 'execution.checkpointing.mode'     = 'EXACTLY_ONCE';

计算层调优要点

1. 资源模型：并行度 = 分区数

湖仓引擎的并行度最终由“数据被切成多少片”决定，两套公式必须烂熟于心：

【Spark】
总核数 C   = num-executors × executor-cores
shuffle 并行度(spark.sql.shuffle.partitions) ≈ (2 ~ 3) × C
Executor 内存 ≈ executor-cores × (4 ~ 8 GB) + memoryOverhead

【Flink】
总 slot = taskmanager 数 × 每 TM slot 数
作业并行度 P ≤ 总 slot（同作业算子可共享 slot，网络缓冲 ≈ P^2 级增长）

【Iceberg 写入侧（Flink sink）】
每次提交文件数 ≈ sink 并行度 × 活跃分区数
单文件大小 ≈ 间隔内写入量 ÷ (sink 并行度 × 活跃分区数)
→ 文件数与大小同时达标的调节旋钮只有两个：checkpoint 间隔 与 sink 并行度

2. 内存配置经验

# Spark：堆内三分（执行/存储/用户）+ 堆外一块
spark.memory.fraction=0.80             # 执行+存储 占 (堆-300MB) 比例
spark.memory.storageFraction=0.30      # 存储下限；缓存重的作业调高
spark.executor.memoryOverhead=4g       # 堆外：Arrow/Iceberg/网络缓冲

# Flink：进程总内存一切从简，让框架自己切
taskmanager.memory.process.size=8g     # TM 进程总内存（推荐显式锁定）
taskmanager.memory.managed.fraction=0.4  # RocksDB / 排序 / 缓存托管内存
taskmanager.memory.network.max=1g      # shuffle 网络缓冲上限
# 经验：TM 堆 2-4GB/slot、单 slot 2GB 起步；状态大就上 RocksDB + managed

3. Shuffle 与网络

# Spark：shuffle 写缓冲与拉取
spark.shuffle.file.buffer=1m                # 每分区写缓冲，大分区表调大
spark.reducer.maxSizeInFlight=96m           # 每个 reduce 任务的拉取总量
spark.shuffle.io.maxRetries=10              # 网络抖动重试，避免整个 Stage 失败
spark.shuffle.io.retryWait=15s
spark.maxRemoteBlockSizeFetchToMem=512m     # 超过此值的远端块直接落盘防 OOM

# Flink：网络缓冲
taskmanager.memory.network.min=256mb
taskmanager.memory.network.max=1gb
taskmanager.network.numBuffersPerChannel=2  # 反压时可下调换内存

4. Checkpoint 大小控制

“Checkpoint 慢 → 反压 → 数据延迟”是实时链路最常见的事故链。控制手段按优先级：

# flink-conf.yaml
state.backend: rocksdb                          # 大状态必选
state.backend.incremental: true                 # 增量快照：只上传变化部分
state.backend.rocksdb.memory.managed: true      # RocksDB 内存托管，防各 TM 自行膨胀
execution.checkpointing.aligned-checkpoint-timeout: 5 min  # 对齐超时自动转非对齐

-- SQL 作业：能遗忘的状态尽早遗忘（如 24h + 1h 兜底）
SET 'table.exec.state.ttl' = '25 h';

• 状态瘦身：窗口长度、去重集合、join 间隔都直接进状态；TTL 是第一杠杆；
• 减少状态化算子：能下推到 Iceberg upsert 的“记录级合并”就不要在 Flink 里开大窗口聚合；
• 增量 + 异步：RocksDB 增量快照 + 异步 barrier 处理，让 Checkpoint 对吞吐的影响降到 5% 以内；
• 观测兜底：Checkpoint 持续时间、对齐时间、状态大小三个指标必须告警（见可观测一章）。

适用边界

“双引擎”不是教条，选型要回到业务延迟与状态复杂度本身：

还有两个“不要用它们硬扛”的边界：交互式亚秒级点查交给 Trino/Doris（见查询层一章）； 高并发在线服务交给 OLAP/KV 引擎（判断标准见存储层 · 适用边界）。 湖仓的优雅之处正在于此——引擎各就各位，数据只有一份。