基于Delta Lake构建数据湖仓体系

直播回放地址：https://developer.aliyun.com/live/249789

导读： 今天很高兴能与大家分享如何通过 Delta Lake 构建湖仓架构。

全文将围绕以下四个部分展开：

• Delta Lake 的基本概念和发展历程，以及 2.0 版本的关键特性
• Delta Lake 的内核解析以及关键技术
• 围绕 Delta Lake 湖格式的生态建设
• Delta Lake 在数仓领域的经典案例

Delta Lake 及 2.0 特性

关于数据湖，数仓以及数据湖仓的概念已经在很多文章及分享中介绍得比较多了，相信大家也都有所了解，在此就不过多重复了，让我们直接来看由 Databricks 提出的数据湖仓 Lakehouse 的关键特性有哪些。

• ACID 事务。一张表可以被多个工作流来读写，事务可以保证数据的正确性。
• Schema Enforcement 和数据管理。Schema Enforcement 也可称作  Schema Validation，在数据写入时，检验数据的 schema 是否能被表所接受，从而来保证数据质量。同时，我们还会对表做一些管理运维操作。
• 支持 BI。湖仓中存储的数据可以直接对接到 BI 系统进行数据分析。
• 支持结构化、半结构化、非结构化数据。数据湖仓提供了统一的、中心化的存储，能够支持各类型的数据。
• 开放性。使用开放、开源的存储格式，如 Parquet 和 ORC 等作为底层的存储格式。
• 支持多类  API。除了 SQL 以外还可以支持如 dataframe 或者机器学习的 API，用以解决 SQL 无法实现的场景。
• 批流一体。简化流式和离线两条数据 ETL 链路，同时降低存在的管理和运维成本。
• 存算分离。每个公司和团队都会关心成本问题。存储和计算分离，按需伸缩，可以更好地实现成本管控。

如上，我们可以发现湖仓大部分的特性是由湖格式来承载和支持的，这就是当前 Delta  Lake，Iceberg 和 Hudi 能够兴起的主要背景和原因。

我们接下来看一下 Delta Lake 的功能迭代和发展历程。

如图上半部分是社区近年的发展，下半部分是 EMR 在 Delta Lake 上的一些进展。我们来介绍几个关键点，首先在 2019 年 6 月份 Databricks 将 0.2 版本作为第一个 release 版本，2020 年的 0.6 和 0.7 版本分别是 Spark2 的最后一个版本和  Spark3 的第一个版本，并从 0.7 版本之后开始支持 DML SQL 的语法，今年（2022 年）的 1.2 和刚刚发布的 2.0 版本放出了比较重大的一些新特性。

阿里云 EMR 对 Delta Lake 的跟进是比较早的，我们从 2019 年就实现了一些比较关键的特性，包括常用 SQL 的覆盖，Z-Order 和 data skipping 的能力，同时我们也逐步解决了 Metastore 同步的问题，实现了无缝与其他产品的访问。Time Travel 也是我们在 Spark2 上就较早支持的功能，目前社区是在 Spark3.3 之后才开始支持 Time Travel 的特性。EMR 这边也提供了自动的 Vacuum 和自动 Compaction 的能力。在数仓场景支持上，EMR 提出了 G-SCD 方案，通过 Delta Lake，借助其 Time Travel 的能力在保留原表结构上，实现了 SCD (Slowly Changing Dimension) Type2 的场景，同时 EMR 也支持了 Delta Lake CDC，使得可以将 Delta 表作为 CDC 的 source 实现增量数仓。

我们再来介绍一下大家比较关注的 Delta Lake 2.0 的一些关键特性。

• Change Data Feed
• Z-Order clustering
• Idempotent Writes
• Drop Column
• Dynamic Partition Overwrite
• Multi-Part Checkpoint

Change Data Feed 和 Z-Order 这些比较重要的特性我们会在之后再重点介绍。

这里着重介绍 Drop Column，它可以结合 1.2 版本发布的一个特性 Rename Column 一起来讲，这类的 Schema 演化都是依赖于 Column-Mapping 的能力。让我们先通过对比 Add Column 和 Change Column 来思考一下，数据写入了 Delta 表之后，Delta 保存了 Schema 信息，同样 Parquet 层面也会保留相同的 Schema 信息，两者是完全一致的，都是以字段名来做标识和存储的。在这样的实现下，单纯的 drop column 还是可以实现的，但是如果在 drop 之后紧跟着 add 一个同名的 column 会是如何呢？这就需要我们在 Delta Schema 和 Parquet Schema 之间做一层映射关系，将每一个字段都能映射到一个全局唯一的标识符，而 Parquet 则保存这些唯一的标识信息。这样的实现下，当我们进行一个重命名列的操作，就可以转化成了对 mapping 配置的修改。

Dynamic Partition Overwrite 只是一个社区一直没支持的语法，大家都比较了解不过多解释。Multi-Part Checkpoint 是用来提高元数据加载效率的特性，Checkpoint 具体是什么我们接下来会进一步讨论。

Delta Lake 内核剖析及关键技术

1. Delta Lake 文件布局

Delta Lake 的元数据由自身管理，不依赖于 Hive Metastore 这样的外部元数据存储。图中的介绍文字分为绿字和橙字两部分，下部橙色标识的是普通的数据或者目录文件，它和普通表没有什么区别，也是以分区的结构管理。区别在于上面元数据的部分，这部分有三类文件，第一类是 json 文件，它记录的是每次 commit 之后产生的信息，每次 commit 会生成一个新的 json 文件；第二类是 checkpoint.parquet，它是由前一次的 checkpoint 文件及其之后的 json 文件合并而来，它的作用是用于加速元数据解析；第三类是 _last_checkpoint 文件，它存储的是上次 checkpoint 的版本号，来快速定位到需要读取的 check point 文件，可以看到前两类文件是元数据的核心。

接下来让我们深入了解 Delta Lake 元数据的组成。

2. Delta Lake 元数据——元素

首先我们来介绍一下基本概念，一张表通常是由两部分组成的，一部分是数据，一部分是元数据。元数据通常存储在 Hive Metastore 中，数据存储在文件系统上。Delta Lake 表也是一样的。它与普通表的区别在于它的元数据是自己管理的，和数据一起存放在自己的文件系统的目录下；另外表路径下存放的数据文件并不是全部有效的，我们需要通过元数据来标出来哪些数据文件是有效的，哪些是无效的。Delta Lake 所有的元数据操作都被抽象成了相应的 Action 操作，也就是所有表的元数据都是由 Action 子类实现的，让我们看目前都有哪些 Action：

• Metadata：保存表的 Schema，Partition 列，及表配置等信息。
• AddFile：commit 中新加入的有效的数据文件。
• RemoveFile：commit 中删除标记为无效的文件。
• AddCDCFile：commit 中新增的 CDC 文件。
• Protocol：Delta 读写协议，用来管理 Delta 不同版本的兼容问题。
• CommitInfo：记录 commit 操作的统计信息，做一些简单的审计工作。
• SetTransaction：存储 Streaming Sink 信息。

3. DDL/DML 组织

在认识了 Action 的元素之后，我们就需要知道不同的操作会对应到哪些 Action 集合，我们以图中的几个例子来说明。首先我们可以看到表中的所有操作都会生成 CommitInfo，它更多起到的是审计作用，而没有实际作用。

接下来让我们看具体操作：

• Create Table。由于只是定义了表，那么仅使用 Metadata 来保存表的元数据信息即可。
• CTAS（Create Table As Select） 。在创建表的同时会加载数据，所以同时会有 Metadata 和 AddFile 的 Action。
• Alter Table。除了 Drop Partition 以外其他的 Alter Table 操作都只是修改元数据，所以这里也只需要修改 Metadata。
• Insert/Update/Delete/Merge。 在没有涉及到 Schema Evolution 的 DML 情况下不会修改元数据，因此不会有 Metadata。我们以 Update 为例，Delta Lake 会先读取 Update 语句中 where 条件可能涉及到的文件，加载这部分数据，然后使用 Update 语句中 set 部分修改掉需要修改的部分，然后连同原文件中不需要修改的部分一起重新写到一个新的文件。这就意味着我们会把读取的文件标记为老文件，也就是 RemoveFile，而新的写入文件我们会使用 AddFile 来标识。

4. 元数据加载

接下来我们来看一下基于 Action 元素如何构建一个表的快照。

首先尝试寻找 _last_checkpoint 文件，如果不存在就从 0 号 commit json 文件读到最新的 json 元数据文件；如果存在，会获取 last_checkpoint 记录的版本号，找到这个版本号对应的 checkpoint 文件，及其之后版本的 commit json 文件，按照版本依次解析元数据文件。通过图中的 6 条规则得到最新的 snapshot 的元数据。最终我们会得到一个最新的 Protocol，Metadata 与一组有效的 AddFile，只要有了这三个我们就知道了这个表的元数据和数据文件，从而组成当前的一个比较完整的快照。

5. Delta Lake 事务

ACID 事务是湖仓一个比较重要的特性，对于 Delta Lake 来说它的 ACID 事务性是以成功提交 json 文件到文件系统来标识此次 commit 执行成功来保证的，也就是对于多个并发写入流能首先将 json 文件的某个版本提交到文件系统的就是提交成功的流。如果大家对存储有所了解，就可以看出来 Delta Lake 事务的核心依赖于数据所在的文件系统是否具备原子性和持久性。这点让我们稍作解释：

• 一个文件一旦被写入，一定是一个完全可见或者完全不可见的，不会存在读取正在写入的不完整数据文件。
• 同一时刻只有一个写入端能够完成对某个文件的创建或者重命名操作。
• 一个文件一旦被写入，后续的 List 操作是一定可见的。

对于并发控制协议，Delta Lake 采用的 OCC，关于该协议的具体原理就不过多展开了。

对于冲突检测，Delta Lake 是支持多个流的同时写入的，这也就造成了势必会有冲突的可能。让我们以一个例子来说明。假设用户当前已经读到了版本号为 10 的文件，并且想将更改后的数据向版本 11 提交，这时发现已经有其他用户提交了版本 11，此时就需要去检测版本 11 与自己提交版本信息是否用冲突。检测冲突的方式在于判断两个提交之间是否操作了相同的文件集合，如果没有就会让用户尝试提交为版本 12，如果版本 12 在这个过程中也被提交，那么对继续检测。如果有冲突的话，会直接报错，判断当前的写操作失败，而不是强行写入造成脏数据的产生。

6. Z-Order

Z-Order 是大家目前比较关注的一个技术。它是一个存在较早的概念，即一种空间的索引曲线，连续且无交叉，能够让点在空间位置上更加聚集。它的核心能力是能够实现多维到单维的映射关系。

接下来我们以一个例子说明。如图所示有 X，Y 两列，X，Y∈[0，7]，图中所展示的是 Z-Order 的排序方式，将数据分为了 16 个文件。对于传统排序来说，如果我们先对 X 再对 Y 做线性排序的话，我们会发现与 X 更靠近的元素会分到一个文件。例如， X 为 0，Y∈[0，3]，4 个竖着的元素将存储在一个文件中，这样我们也就同样可以生成 16 个文件。这时如果我们想查询 4<=Y<=5，这时就需要我们将全部扫描下半部分的 8 个文件（竖着的 4 个元素为一个文件，Y∈[4,7]）。如果我们使用 Z-Order 来排序的话可以看到我们就只需要扫描四个文件。

让我们再举一个例子，如果我们要查询 2<=X<=3 and 4<=Y<=5，如果按照 Z-Order 来排序的话就只要扫描一个文件，按照传统线性排序的方式需要扫描 2 个文件（X∈[2,3],Y∈[4,7]）。可以看到在我们使用 Z-Order 之后需要扫描的数据量减少了一半，也就是说在同等计算资源的情况下我们的查询时间可以减少一半，使性能提升一倍。从以上例子我们可以看出线性排序更关注的是当前排序的字段的聚集效果而不是空间的聚集效果。

7. Z-Order+Dataskipping

Z-Order 只是帮我们做了一个文件布局，我们要结合 data skipping 才能发挥它真正的效果。这两个在功能上是各司其职互不干扰的，它们没有任何的在功能上的耦合，但是它们却必须是需要相互辅助的。我们可以想象一下，如果没有 Z-Order 这样的拥有良好聚集效果的文件布局，单独 data skipping 是不能实现较好的文件过滤效果的，同样只有 Z-Order 没有 data skipping，其单纯的文件布局也起不到任何的读取加速的作用。具体的使用过程是：在写入时，完成对数据 Z-Order 排列，写入文件系统，并以文件粒度提取文件对应字段的 min-max 值，写入如图所示的 AddFile 的元数据 stats 中。在查询时，使用 min-max 值做过滤，选出符合查询条件的需要加载的文件，之后对数据再做过滤，从而减少文件和数据的读取。

这里有一个需要关注的点是如果当查询模式改变了，比如说原来是基于 a，b 两个字段做 Z-Order，但是一段时间之后主要查询的是 c 字段，或者文件经过了多次写入，它的聚集效果都会产生退化，这时就需要我们定期重新执行 Z-Order 来保证聚集效果。

Delta Lake 生态建设

上面我们提到了 Delta Lake 的一些基本概念，大家也可以看到基于目前的大数据架构我们没法通过一个单一的系统来构建整体的大数据生态，接下来我们就来了解一下 Delta Lake 目前的生态系统如何辅助我们搭建大数据体系。

首先我们来看开源的生态。对于大数据组件我们可以粗略地分为存储、计算以及元数据管理。元数据管理的事实标准是 Hive Metastore，存储主要有 HDFS 及各云厂商的对象存储，各种计算引擎都有相对应的存储接口。对于查询来讲，由于各个引擎的框架语义或者 API 的不同会导致每个湖格式都需要和查询/计算引擎一对一的对接支持。

我们结合几个典型的引擎来介绍一下目前的开源生态。

Delta Lake 本身就是 Databricks 公司开源的，所以它们对 Spark 的支持从底层代码实现到性能上的表现都比较好的，只不过对于开源版本来说某些 SQL 功能还没有完全开放或支持。阿里云 EMR 的 Delta Lake 版本目前已经覆盖了常用 SQL。

对于 Hive，Presto 和 Trino 来讲目前社区已经实现了查询的功能，写的能力目前还不支持。以上三种引擎接口的实现都是基于 Delta Standalone 项目来实现和拓展的，该项目内部抽象了一个 Standalone 的功能来对接非 Spark 的计算和查询引擎的读写功能。

这里提几个目前社区还没有很好支持的点：

• 通过 Spark 建表之后，是不能直接使用 Hive 等引擎去查询的，需要用户在 Hive 侧手动创建一个外表才能做查询。其原因是 Hive 查询 Delta 表需要通过  InputFormat 来实现，而 Spark 侧创建的 Delta 表在将元数据同步到 Hive Metastore 时，没有获取到正确的相关信息（其他表类型如 Parquet 和 ORC 等是在 Spark 源码内硬编码到了 HiveSerde 类中），也就没法实现正确的元数据同步。我理解这个主要原因是 Spark 没有考虑到这些场景，实现比较好的拓展能力，同时 Delta Lake 社区也没有想将同步元数据相关的逻辑嵌入到其代码实现中。

• 在 Hive 中创建 Delta 外表不能指定分区字段，即使本身 Delta 是一个分区表，对于 Hive 引擎来言也将其视为普通表。这里提一点，这样的设计并不会引起性能的差异，Delta Standalone 内部依然会根据查询条件进行分区裁剪。

对于以上两点，阿里云 EMR 已经做了比较好的支持：使用 Spark 建表会自动同步元数据到 metastore，然后直接通过 Hive，Presto，Trino 去查询，不需要任何额外的操作。同时我们支持将表的分区特性正确的显示在 Hive Metastore 中，避免用户使用时的困惑。

另外，在 Hive 等基于 standalone 模块实现的查询引擎上查询 Delta 表会存在元数据加载效率问题。如 Hive 查询，Delta 表的元数据加载是在 Hive CLI 本地去完成的。元数据比较大的情况下，会占用大量的内存和时间。在 EMR 上我们实现了 emr manifest 元数据加速的能力，在每次写入时将最新快照关联到的 AddFile 信息提前写入到文件系统中，查询时跳过元数据加载来解决该场景下的元数据加速问题。

同时我们在 Presto/Trino 上支持了 TimeTravel 查询和 dataskipping 优化。

最后对于 Flink 的写入，Delta 在 0.4 版本开始社区发布了 Flink sink 的功能，在 0.5 发布了 Flink source 的功能。

接下来我们来介绍一下阿里云生态对 Delta Lake 的支持。我们目前已经实现了 Dataworks，MaxCompute，Hologres 对 Delta 表的查询；对接并支持使用阿里云数据湖构建 DLF 作为元数据，助力实现更好的湖仓一体。同时我们也对接了 DLF 的湖表自动管理模块，这点我们展开介绍一下。

在湖格式中我们引入了版本的概念和批流一体的功能，这会造成有一些历史版本的数据在当前的快照下已经失效，或者在流式场景下产生一些小文件，再加上我们刚刚提到的 Z-Order 的效果会随着时间退化，这些问题都需要我们对湖表进行一些管理的操作，如我们需要定期做历史文件的清理，重新执行 Z-Order，以及做一些文件合并的操作等。DLF 这里我们实现了自动化的湖表管理模块，会实时感知表的版本更新，实时分析表的状态（如有效文件占比、平均文件大小等指标），结合策略中心预定义的策略来采取相应的操作，透明地帮助用户实现表的管理。同时我们也拓展了对湖表生命周期的管理，对于一些老的分区如果我们使用频率较低我们可以对其进行压缩或者移到低成本的存储中去。同时 DLF 的 data profiling 模块也会实时统计表级别或者分区级别的各个维度的统计信息，更新到指标库，用于进一步的查询加速或者湖表管理等。

Delta Lake 经典数仓案例

最后我们来看一下 Delta Lake 经典数仓的案例。

Slowly Changing Dimension（SCD，缓慢变化维），SCD 是用来解决在数仓场景中随着时间缓慢变化的维度数据的。根据对变化之后的新值的处理方式，定义了不同的 SCD 类型，这里我们着重讨论一下 Type2：通过新增一行记录的方式来保存历史值的这种类型。通常情况下我们在传统数据库内我们首先会在表中添加 Start 和 End 列来标识当前维度值的生效范围，如果 End 值为空表示当前维度在最新版本是生效的，我们也可以再添加一列状态列来表示当前维度值是否生效。更多的时候我们不会关注每一次变化，而只关心一个固定的业务周期或者一个时间段内最新的值是什么。举一个例子：我们将用户和其所在地做成的一个维度表，假设用户 A 从北京迁到杭州、武汉，在表中用户 A 不同时间就会有不同的地址。我们想知道 2022 年 7 月 16 号用户 A 的所在地，也就是它最终的所在地武汉，而不是关注用户 A 早上从北京到了杭州，中午又从杭州到迁了武汉的过程。

SCD Type2 的传统方案大致如下：通过实时流不断获取增量数据写入到增量表中，当 T+1 的数据全部处理完后我们会和离线表的 T 分区做合并，从而生成离线表的 T+1  分区。在使用过程中我们只需要基于离线表，通过分区字段来指定到一个固定的粒度（如天）去查询相关数据。这里存在的缺点是离线表的 T 和 T+1 数据时高度重复大量冗余的，这就造成了很明显的存储浪费，同时离线和事实两条工作流也增加了管理和运营的成本。

那么我们来看 Delta Lake 是如何解决以上问题的。刚才提到了我们更关心的是一个固定时间段内的最新值，所以我们将其命名为 G-SCD——基于固定粒度的缓慢变化维。Delta Lake 这样的湖格式是具备多版本的概念的，那么就可以利用 Time Travel 的能力查询到历史的某一个快照的数据，同时保障查询性能和数据不重复存储，EMR G-SCD 就利用上以上的特性来进行构建，让我们来看具体的解决方案：

首先 MySQL 会将 binlog 同步到 Kafka，之后会由 Spark Streaming 来消费，最终我们会将数据提交给 Delta Lake。

整个流程看着和普通的流式写入没有什么区别，但关键在于：

① 最终将数据和业务快照信息一起提交。

② Spark Streaming 会对 batch 数据按照业务快照进行切分，保证每次提交的仅包含一个业务快照内的数据，同时会将已经处理完的快照做 save point 来永久保留某版本。

在 G-SCD 的实现上有两个核心的问题要解决：

① 业务快照与 Delta 版本之间的映射。如图所示，通过每次 commit 关联到一个具体的业务快照（Delta 版本 V7 和 V8 提交的都是业务快照 T 的数据），并且要求业务快照随着 Delta 版本递增（从 T-1，到 T，再到 T+1 业务快照）。这样就可以将查询某个业务的快照例如 7 月 15 号的数据，映射转化成对某一个具体的版本的 Time Travel 去做查询。

② Savepoint&Rollback。对于传统方案来讲，只要不主动删除分区，分区是不会丢失的，而湖表则具备自动化清理历史版本的能力。G-SCD 方案下我们并不是需要保留所有的版本，而是希望能够指定某一个版本能够保留而不被删除，所以在这里我们需要 save point 的功能。另外的一点是数据难免是有错误的，我们也就需要版本回溯的功能，能够回溯到某一天的数据从而重新修补数据。这里 rollback 的功能相当于社区  2.0 版本发布的 restore 功能。

对流式数据处理比较熟悉的同学能会发现这里存在数据漂移的问题，这个现象产生的原因就是前一个快照的数据到了下一个快照周期才到，那么这个情况下我们该怎么处理？G-SCD 会要求业务快照在 Delta 版本上是递增的，这点已经提到。同时该方案也会要求上游 Kafka 的 Partition 是按照业务快照严格有序的，同时同一个 ID 的数据只能落到同一个 partition 内，这样在处理某一主键的数据上就永远不会出现错序的情况。之后在 Streaming 的层面上我们会判断每一个 batch 是否属于同一个业务快照，如果是的话就直接提交，如果不是的话我们就仅仅提交业务快照周期小一点的数据，而将另一部分数据先做缓存。对于缓存机制，我们会先将首次出现的下一个快照数据暂存，先去处理由于合理数据漂移的前一个快照数据。在一定时间之后，当我们认为不会再有漂移数据的情况下我们才会将这部分数据提交。通过这样的切分可以保证 Delta 侧每一个 commit 只会对应到一个业务快照的数据。

接下来让我们看 G-SCD 方案所具备的优点：

• 批流一体，降低管理成本
• 充分节省存储资源
• 充分利用 Spark/Delta 的查询优化
• 不需要像其他 SCD Type2 的实现方案那样添加多个辅助字段；同时保留了传统方案中使用 dt 作为分区方式，从而可以复用原有 SQL，使用户迁移无成本。

该方案已经被阿里云的客户广泛地应用到了生产实际当中。

Change Data Capture（CDC, 变化数据捕捉）。最后让我们来讲一下 CDC 场景，这里涉及到 Delta 2.0 发布的非常重要的 CDF 特性。CDC 是一个用来捕捉和识别数据的变化，并将变化的数据交给下游来做进一步处理的场景。CDF 是一种能让表或者数据库能够具备吐出变化数据的能力。CDF 的输出结果可以标识出数据做了什么样的变化，比如 insert，update 或者 delete，以及可以让我们知道数据在变更前后的内容，CDF 同时也包含了版本数据变化的时间点和版本号信息。在 Delta Lake 中开启 CDF 只需将 delta.enableChangeDateFeed 设为 true。

在没有 CDF 之前，一般我们只能通过 MySQL 取 binlog 的形式实现到 ODS 层的增量数据更新，但是到下游 DWD，DWS 层我们只能通过低效的全量的方式去做数据更新了。当我们具有可 CDF 的能力之后我们就能够实现将湖格式作为 CDC 的一个源实现从 ODS 到 DWD，DWS 的全链路的增量实时数仓。接下来让我们来看一个具体的案例。

如图我们定义了一个数据源和三张表。user_dim 是一张维表，user_city_tbl 表示用户所在地，city_population_tbl 用来统计城市常驻人口。user_city_tbl 表的更新需要 source 源和 user_dim 表做 join 后写入。city_population_tbl 表是通过对 city 字段做聚合产生的。现在让我们将两张表都开启 CDF，看一下会有什么数据产生。比如当前在上游来了两条数据，user1 来自杭州，user5 来自武汉，通过 Merge 语句将数据加载到 user_city_tbl 中，如图，user1 已经存在所以会更新地址信息，user5 为新用户所以插入数据。对于更新操作会有两条数据来表示，一条是 pre_update，表示更新前的旧值，一条是 post_update，表示更新后的新值。对于新插入的数据我们只要一条数据来表示插入操作，没有旧值。对于删除操作，CDC 当前的值表示它的一个旧值，没有新值。

可以看到这里输出的格式采用了一种不同于大家比较熟悉的 MySQL binlog 或者 debezium 的格式，相比较而言，CDF 的实现方案对下游去做数据处理是更加友好的，它同时也包含了我们所需要的所有的信息，不需要做过多转换。如果我们使用 binlog 或者 Debezium 的话还需要从 json 字符串中提取出我们需要的列的信息。

使用 user_city_tbl 的 change data 对下游 city_population_tbl 做增量更新，最终实现对 city_population_tbl 表中 bj 城市的人口数减一，对 hz 和 wh 的城市人口数加一。从这里我们也看以看出 CDC 的输出数据是需要包含 update 或 delete 数据的旧记录的详细信息的，不然就无法增量更新 bj 城市的人口数，准确的实现数据聚合的操作。

流程继续，如果 city_population_tbl 表也需要用做 CDC source，开启 CDF 之后的 CDC 输出信息如右下图所示。

最后再让我们看通过 Delta Lake 实现 CDC 的设计和实现。

Delta Lake 通过 CDF 方案来实现 CDC，其理念是在必要场景下持久化 CDC 数据，尽可能地复用已有的数据文件，来充分平衡读写两端。

不同于一些传统数据库它们有自己的常驻服务，可以在不影响写入效率的情况下直接后台生成相关的数据，Delta Lake 仅仅作为数据存储层的数据组织方式，数据读写的执行还是依赖于计算引擎本身，比如 Flink 或 Spark，其所有额外的开销也需要在当前 commit 完成，从而会影响写入效率。

之所以不采用完全依赖于查询时通过类似版本间 Diff 或者 Join 的方式来实时计算出 Change Data，当然考虑的是查询性能。在这里通过一个场景来明确一下 CDC 的一个可能被忽略的点，即 CDC 需要感知到每一次相邻 commit 间的变化，而不能仅仅是查询方位内首尾两个 commit 的变化。湖格式 CDC 是基于单 commit 来讲的，也就是说如果有一条数据，第一次 commit 从 1 变到了 2，第二次 commit 从 2 变到了 3，那么这两次 commit 的 CDC 数据应该是从 1 到 2 再到 3，而不是直接由 1 到 3，部分 CDC 的实际生产场景要求这样的能力。

在设计方案上 Delta Lake 提供了仅在无法简单的通过当前 commit 信息获取完整数据变更时才会持久化 CDC 的能力，这里完整的 CDC 包含前值和新值，包含所有的操作以及时间戳和版本号的信息。这意味着可以直接读取和加载 CDC 数据而不需要通过读历史的快照数据来计算得到。

在了解了以上 CDF 设计特点之后我们会发现，有一部分场景需要持久化 CDC，另一部分场景不需要持久化 CDC。我们先来聊一下不需要持久化 CDC 的场景，也就是哪些操作可以通过当前的 commit 信息直接返回 CDC 数据。这里举两个例子，第一个是 Insert into，Insert into 语法新增加的 AddFile 不会对其他的数据有任何影响，其相关的元数据 commit json 文件中的元数据只有 AddFile，所以我们可以直接加载这些 AddFile 文件的数据，对每一条记录加上 insert 的操作标识，同时添加 timestamp 和 version 信息，转换成 CDC 的格式返回即可。第二个例子是 Drop Partition，这个功能在社区是没有支持的，在阿里云 EMR 上支持。它会将某一分区下的所有有效数据都标识为 RemoveFile，当我们读取 commit json 文件时我们得到只有 RemoveFile 的文件列表，那么我们就可以加载 RemoveFile 标识的数据文件，对于每一条数据添加 delete 的操作标识，并且加上 timestamp 和 version 信息。对于类似这样的操作，CDF 的实现方案没有增加任何的写入开销，直接复用已有的数据完成，加载转换得到 CDC 数据返回。

那我们再来看一下哪些是需要持久化 CDC 的。如 Update 操作，需要将某个数据文件中的部分数据更新后，连同未更新的部分一起写入到一个新的数据文件。这样的场景下，就需要将更新的部分数据直接转化成要输出的 CDC 格式数据，然后保存到文件系统。在查询时，对于这样的 commit，直接去读取它包含的 CDC 文件，加载返回 CDC 数据。持久化 CDC 数据的文件，就是通过刚才未详细解释的 AddCDCFile 这个 Action 来记录的。

如上图，CDF 方案下持久化的 CDC 写入到单独的 _change_data 目录下（图中红色部分）。