关于后端:基于Delta-Lake构建数据湖仓体系

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

导读： 明天很快乐能与大家分享如何通过 Delta Lake 构建湖仓架构。

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

Delta Lake 的基本概念和倒退历程，以及 2.0 版本的要害个性

Delta Lake 的内核解析以及关键技术

围绕 Delta Lake 湖格局的生态建设

Delta Lake 在数仓畛域的经典案例

对于数据湖，数仓以及数据湖仓的概念曾经在很多文章及分享中介绍得比拟多了，置信大家也都有所理解，在此就不过多反复了，让咱们间接来看由 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 具体是什么咱们接下来会进一步探讨。

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 目前的生态系统如何辅助咱们搭建大数据体系。

首先咱们来看开源的生态。对于大数据组件咱们能够粗略地分为存储、计算以及元数据管理。元数据管理的事实标准是 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 经典数仓的案例。

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 目录下（图中红色局部）。

关于后端:基于Delta-Lake构建数据湖仓体系

Delta Lake 及 2.0 个性

Delta Lake 内核分析及关键技术