同一任务，两大引擎：InnoDB 与 Iceberg 内部机制

# 同一份工作，两台引擎：InnoDB 与 Iceberg 的内部机制 原文：https://marcpaul.io/logs/same-job-two-engines-innodb-and-iceberg-internals/ 我的 StockAlgo 分析仓库（https://github.com/MarcPaulTheCoder/StockAlgo）的第一版在 OLTP 引擎上运行分析，第二版则在 OLAP 技术栈上运行相同的工作。我选择 OLAP 技术栈是因为一项单一的优化，但为了利用它，我不得不了解关于数据湖仓的一切。 StockAlgo v1 的六张主要表的行数，取自我家庭实验室中的 StatefulSet。 ## OLTP ### InnoDB 的三大特性 **行存储将所有列放在一起** 获取某个标的的一根 K 线只需触及磁盘上的一个位置，这正是数据摄取所需要的。即使一个指标只需要两列数据，它也会通过缓冲池读取所有列。 **索引加速读取，减慢写入** 每个索引都是一棵独立的 B-Tree，每次写入都必须更新表中的所有 B-Tree。B-Tree 的读取速度极快。任何索引的目标都是以牺牲写入性能为代价来加速读取。这非常适合选择性读取，但对于写入繁重的摄取工作负载来说则非常糟糕。这也是我让 `Raw*` 表只保留主键的原因，以及为什么在接收 `Raw*` 表数据的表上部署了大量索引和分区。 **Undo 日志支持并发读写** 当一行被更新或删除时，旧值会存储在一个叫作 undo 日志的结构中。如果某个会话在写入期间读取该表，旧值会从 undo 日志中获取。插入操作也有一个较小的 undo 日志，其中存储了被插入的表和主键信息，这样 InnoDB 就知道如果插入未提交时该回滚什么。插入提交后，其 undo 日志会被立即清除。更新或删除的 undo 日志则会加入一个叫作**历史链表（history list）**的全局列表，正是它让会话能够读取被修改数据的旧版本。对于更新和删除繁重的数据库，历史链表会快速增长，处理起来将是噩梦。幸运的是，比我更聪明的人已经造好了解决方案，你猜对了，它就内置在 InnoDB 中。InnoDB 会标记 undo 日志以便清除，条件是在提交之前或提交期间开始的所有会话都已完成。这套系统叫作**多版本并发控制**，简称 MVCC。抛开所有技术术语不谈，用大白话说，MVCC 让插入和更新能够在不阻塞读取的情况下进行。 在不知不觉中，我按照 OLAP 的设计原则设计了我的 InnoDB 数据库。`financialdata` 表存储了 `datetime`、`open`、`high`、`low` 和 `close`。它是唯一持有 `datetime` 的表。任何我想查询的技术指标都需要对 `financialdata` 表进行连接（join），才能知道指标所在行对应的日期。这是我最糟糕的数据设计之一。直到数据库超过 100GB 数据之后，我才意识到这种设计行不通。修复它的过程非常痛苦。 ```sql CREATE TABLE `FinancialData` ( `FinancialDataID` int NOT NULL AUTO_INCREMENT, `StockID` int NOT NULL, `StockDate` datetime DEFAULT NULL, `Open` decimal(20,5) DEFAULT NULL, `Low` decimal(20,5) DEFAULT NULL, `High` decimal(20,5) DEFAULT NULL, `Close` decimal(20,5) DEFAULT NULL, `Volume` decimal(20,5) DEFAULT NULL, `EMA` decimal(20,5) DEFAULT NULL, `VWAP` decimal(20,5) DEFAULT NULL, `RateOfChange` decimal(20,5) DEFAULT NULL, `OnBalanceVolume` decimal(20,5) DEFAULT NULL, PRIMARY KEY (`FinancialDataID`,`StockID`), UNIQUE KEY `StockID_Date_Idx` (`StockID`,`StockDate`), KEY `idx3` (`StockID`,`FinancialDataID`,`StockDate`,`Close`) ) ENGINE=InnoDB AUTO_INCREMENT=320075156 DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_0900_ai_ci /*!50100 PARTITION BY HASH (`StockID`) PARTITIONS 100 */; ``` 在 MySQL StatefulSet Pod 内部。`PARTITION BY HASH PARTITIONS 100` 意味着磁盘上有 100 个 FinancialData 的 ibd 文件，每个分区一个。这是负责我的 `financialdata` 表分区的文件。每个 `.ibd` 文件都是一个 InnoDB 表空间文件，包含大约表数据的 1/100。文件范围为：`FinancialData#p#p0.ibd` 到 `FinancialData#p#p99.ibd`。 公平地说，这个设计背后的想法是合理的。`financialdata` 表有 `datetime`、`open`、`close`、`low`、`high`、`StockID` 以及不少其他列，从上面的 schema 可以看到。当我刚开始做这个数据库时，我以为每个查询都需要这些列。说实话，当我开始做这个数据库时，我甚至不知道最重要的价格是收盘价。只能说我进步了很多，我现在连《华尔街日报》都看了，大家都知道这意味着什么。话说回来，这个设计意味着我所有的指标表都包含 `FinancialDataID` 和 `StockID`，它们作为高效的连接列。把 `financialdata` 当作基础表，然后根据我正在使用的指标"挑选"我需要的列，这个想法听起来和只查询我想要的列、忽略表中其他列非常相似。 当我终于鼓起勇气在指标表中添加 `StockDate` 时，我用了老可靠的方法——一个由多个不同的计划事件触发的存储过程。每个计划事件在单独的表上运行存储过程，所以锁不是问题，等待才是。 ```sql CREATE TEMPORARY TABLE IF NOT EXISTS t (StockDate DATETIME, FinancialDataID INT PRIMARY KEY); INSERT INTO t(StockDate, FinancialDataID) SELECT StockDate, FinancialDataID FROM FinancialData WHERE StockID = iStockID; UPDATE sar sr INNER JOIN t on sr.FinancialDataID = t.FinancialDataID SET sr.StockDate = t.StestDate; UPDATE macd md INNER JOIN t on md.FinancialDataID = t.FinancialDataID SET md.StockDate = t.StockDate; UPDATE directionalmovement dm INNER JOIN t on dm.FinancialDataID = t.FinancialDataID SET dm.StockDate = t.StockDate; UPDATE boilerband bb INNER JOIN t on bb.FinancialDataID = t.FinancialDataID SET bb.StockDate = t.StockDate; UPDATE chaikinoscillator co INNER JOIN t on co.FinancialDataID = t.FinancialDataID SET co.StockDate = t.StockDate; DROP TEMPORARY TABLE IF EXISTS t; ``` 每个计划事件只运行其中一个更新语句，其余的都注释掉了。 在 v1 使用了 5 年之后，我意识到我的分析工作每张表只用到 2 到 3 个列，而我被迫读取每个列。既然优化分析工作在理论上可能给我带来市场优势，我决定朝这个方向优化。经过一些研究，我得出结论：引擎选错了。 ## OLAP 如果你觉得 InnoDB 那一节已经够技术了，老兄，那你可要做好被暴击的准备了。 v2 的分析表是基于 MinIO 上的 Iceberg，使用 Polaris 目录服务管理的 Parquet 文件，由 Spark 负责读取。整个分区方案只是一条 DDL 子句，摘自这里（https://github.com/MarcPaulTheCoder/StockAlgo/blob/main/ddl/stockTableCreation.sql）： ```sql CREATE TABLE IF NOT EXISTS raw_bars (`symbol` string, `time_stamp` timestamp, `open` double, `high` double, `low` double, `close` double, `volume` bigint, `vwap` double, `trade_count` bigint, `is_valid` boolean) USING iceberg PARTITIONED BY (bucket(16,`symbol`)); ``` 使用 Iceberg 的一个便利之处在于它会在元数据中计算并存储任何分区信息。这意味着如果我想在分区中添加一列，只需在读取时付出代价。数据按分区版本分组，每个版本独立剪枝，最后合并剪枝结果。这个步骤叫作**分片计划（split planning）**。在 v1 中分区一张表需要几天到一周的时间，我在做 v1 时喝了很多咖啡休息，而如果我想更改分区，就必须重新对整张表做分区，这就是我选择比正常更大的哈希值 100 的原因。约 13,000 个 StockID 被分成每组 130 个 StockID 的分组，这意味着在需要重新分区之前会经历很多年。 ## 读取路径 我花了大约一个月才理解我数据湖仓的查询优化器，因为它是一个相对复杂的主题，而且，它是我为数不多无法与我的 OLTP 背景联系起来的概念之一。我要剧透一些技术细节了，我一直在纠结清单文件（manifest）和数据文件（data file）的区别。就整体而言，这两个定义算是比较简单的，但我就是无法理解数据分组发生在多个不同的层次上。理解清单文件和数据文件就是我的"顿悟时刻"。言归正传，数据湖仓由三个组件组成：Apache Polaris、MinIO 和 Apache Iceberg。 **Polaris** 充当我的目录服务，对我来说，它是一个 Postgres 数据库和一个用于读写的 API。在查询优化方面，Postgres 数据库的主要职责是存储每张 Iceberg 表指向 `metadata.json` 文件的指针。每个查询开始时先请求该指针，每次提交结束时更新它。Polaris 一次只接受一个更新请求，这保证了表更新的原子性，当两个写入者同时出现时会发生什么则在提交路径中介绍。显然，这是我在湖仓中最喜欢的部分，因为它有一个 OLTP 数据库。 **MinIO** 是我湖仓的物理存储。我使用 MinIO 的唯一原因是我在家庭实验室中托管湖仓。MinIO 的主要用途是存储 Parquet 数据文件、清单文件（manifests）和清单列表（manifest lists），以及 `metadata.json` 文件。使用 MinIO 意味着文件结构与 S3 Bucket 完全一致。MinIO 只负责这些——用自制的东西替代真正的东西。 最新文件在前。每个交易日的 22:02 完成一次完整的提交集，缺失的日期是周末，没有 K 线，没有提交。我的 9 月份提交在 22:02 左右写入的实际文件：`metadata.json` = 表的新更新版本，`snap-*.avro` = 清单列表，`m0.avro` + `m1.avro` = 清单文件。由于周六和周日休市，`MERGE` 是空的，所以这些文件不存在。 **Iceberg** 是大脑。Polaris 和 MinIO 各自有非常具体的职责，其余的由 Iceberg 承担。Iceberg 给了我们两层查询优化，我们在数据文件层面还有一层额外的优化。我喜欢把优化层想象成俄罗斯套娃——我们从最小的娃娃开始，把每个娃娃放进下一个最小的娃娃里，从数据文件到 `metadata.json` 文件，逐层构建摘要。 最小的娃娃是新写入的数据文件。数据文件顾名思义，就是表中写成一个叫作 **Parquet 文件** 的数据。Parquet 文件是为列式存储系统优化的，让我们可以选择要读取的列，而不必每次都读取所有列。数据文件是最小的娃娃的原因是每个 Parquet 文件包含不同的行分组（Row Groups），每个分组都有一个摘要，具体来说是每列的最小值/最大值和空值数量。默认情况下，行分组按照到达顺序进行批处理。你可以通过 `WRITE ORDERED BY` 来定义行的分组方式。我没有在我的湖仓中使用 `WRITE ORDERED BY`，因为我很高兴我的湖仓能正常工作，还没回去添加 `WRITE ORDERED BY symbol, time_stamp DESC`。注意，这和我第一版使用的唯一键组合是一样的。 下一个娃娃是我们刚才那个娃娃的更大版本。数据文件被一个叫作**清单文件（Manifest）**的东西分组。就像行分组一样，清单文件跟踪每列的最小值/最大值和空值计数，还跟踪每个数据文件的分区值、文件大小和位置。和行分组相同的概念，只是摘要范围更大。清单文件还负责跟踪一个叫作**删除文件（Delete Files）**的东西。删除文件顾名思义，它们跟踪被删除或更新的行。清单文件将删除文件链接到数据文件，这让 Iceberg 在返回查询结果之前可以删除删除文件中的任何行。如果表存在删除文件，数据文件和删除文件的合并在每次读取时发生。这叫作**读时合并（merge on read）**。虽然在每次读取时计算而不是在一次写入时计算看起来没什么用，但有一个非常流行的用途——流式处理。记下发生了什么变化，之后再处理，比当场做修改要快得多。 最大的套娃是**清单列表（Manifest List）**。和上一步完全一样，清单列表是对清单文件的上一层摘要。清单列表给出了每个清单文件计算出的分区值的最小值/最大值，以及每个清单文件写入时所使用的分区版本。跟踪分区版本使分片计划成为可能。一个清单列表属于一个快照，快照存储指向其关联清单列表文件的指针。快照 ID 和清单列表指针都存放在 `metadata.json` 文件中。表的 `metadata.json` 文件还通过快照 ID 告诉我们表的历史，以及每个快照 ID 对应的清单列表，还有关于表的实际元数据，告诉我们每一列的信息。我保证我没有被赞助（目前还没有），但 Netflix 做软件真的很厉害。 Midnight Commander 在 MinIO 数据路径内，按修改时间排序。每次提交是一个共享同一时间戳的集群，包含一个 metadata.json、两个 manifest avro 文件和一个 snap avro 清单列表，从 5 月 5 日的 CREATE TABLE 开始。呈现在你们面前的是我湖仓的演化史。5 月 5 日有表了！0000 表示这张表是第一个迁入新家的，它孤零零的，没有数据相伴。快进到 6 月 8 日，我们看到 0000 表的后代 0001，它有两个 avro 文件相伴，`*-m0.avro` 充当清单文件，而 `snap-*.avro` 充当清单列表。我们还可以通过追踪 `000**.metadata.json` 文件来了解这个湖仓的可爱谱系。是的，我已经彻底沉迷了…… 与其追溯我的湖仓如何优化查询，不如让我走过一个准备写入的执行器会经历什么。MERGE 必须先找到匹配的行以及它们所在的文件，才能重写它们，所以写入始于读取。执行器从最大的套娃开始，向 Polaris 请求表指向当前元数据文件的指针。元数据文件命名当前快照。快照指向其清单列表（第二大的娃娃），计划器计算我的标的所属的桶，只进入该标的所属的桶。我们只检查分区范围包含我们计算出的标的的清单文件（第三大的娃娃）。在上一轮剪枝中存活下来的数据文件会检查每个行分组的统计信息，以剪枝掉不需要检查的行分组（最小的娃娃），然后解码仍然存活的行分组的数据文件。在湖仓的不同层级上有三层数据过滤。这是大多数列式数据仓库和湖仓背后的一般模型。Snowflake、BigQuery 和 Delta Lake 都运行着某种变体。高层路径是：在写入时标记统计信息，读取任何数据之前进行剪枝，以及列式解码，主要区别在于各厂商之间的命名不同。学会了一个就意味着学会了大部分。 我的指标表有 36 列。当一个指标需要 `close` 和 `volume` 时，我终于可以说那两列被读取了，而其他 34 列留在磁盘上，它们该在的地方。这只花了我大约 8,000 个字和 9 个月。 ## 提交路径 数据文件是不可变的。写入会产生新文件。我的数据摄取通过一个 `MERGE` 完成，摘自这里（https://github.com/MarcPaulTheCoder/StockAlgo/blob/main/src/stockalgo/ingest.py）： ```sql MERGE INTO stockalgo.market.raw_bars AS rb USING new_raw_bars AS nrb ON rb.symbol = nrb.symbol AND rb.time_stamp = nrb.time_stamp WHEN MATCHED THEN UPDATE SET rb.time_stamp = nrb.time_stamp, rb.open = nrb.open, rb.close = nrb.close, rb.high = nrb.high, rb.low = nrb.low, rb.volume= nrb.v