一、数据湖&数仓&湖仓
新架构的提出往往是为了解决老架构存在的问题，而数据湖的出现也是为了解决传统数仓架构遗留的问题。
阿里云对数据湖的定义为：数据湖是统一存储池，可对接多种数据输入方式，可以存储任意规模的结构化、半结化、非结化数据，也可以无缝对接到多种计算分析处理的引擎和平台，可以直接对数据做处理和分析，打破数据孤岛，洞察业务价值等。同时，阿里云还提供了冷热分层转换能力，覆盖数据全生命周期。
从定义可以提炼出数据湖的公共能力：
第一，具备统一中心化的存储和数据管理。
第二，支持非结构化、半结化和结构化等各类型数据。
第三，支持 SQL 、ML等各类查询、计算、分析。
第四，支持大规模的数据。
第五，开放开源的格式，比如Parquet、ORC等。
传统的数仓架构仅支持结构化和半结构化，只能提供 SQL 语法，导致对于音频视频、图等非结构化数据的处理能力较弱，也无法直接对接到机器学习和深度学习框架。
为了解决数仓引入等问题，数据湖架构被提出。而数据湖本身存在一些问题：
第一，安全。数据湖将所有数据都做了统一存储，缺乏完备的操作审计，且只能做到文件级别的权限控制，因此存在安全性问题。
第二，质量。数据湖天然支持同一份数据同时被数据分析师、数据科学家等不同的角色使用，同时还引入了批、流两种方式的写入，如果没有合适的方法保证事务的质量，保证其一致性，则会带来可靠性问题。
第三，性能。不同于数仓可以使用索引、缓存以及优化后特定的存储格式加速查询，原生的数据湖查询性能较弱，
第四，成本。数仓的数据格式是封闭、自定义的，因此在机器在学习和深度学习领域无法很好地应用。
基于以上挑战，数据湖引入了存储层。而引入新的引擎或组件必然将带来一系列学习和使用成本，以及后续的运维和管理成本等。因此，最终从数据湖衍生出数据湖仓架构。
湖仓架构是在数据湖上拓展了原本仓的能力，融合了两者的优势，弥补劣势，达到了灵活性、安全和高性能的统一。
湖仓架构的底层继承了数据湖架构统一存储数据的能力，使其能够承载结构化和非结化的数据，同时允许 BI 和 AI 的任务直接在上面运行。湖仓架构的关键在于中间层封装了数据湖的存储层，包括数据湖管理和元数据服务的能力。
通过权限、审计工作解决安全问题以及降低部分数据湖管理成本的工作，可以借助中间层的数据湖管理和元数据服务模块。性能和质量上多流读写的事务性保证可以借助于数据湖格式本身来解决。
另外，数仓架构能够支持数据湖的数据增加或删除字段等schema 变更来适配数据的变更，支持数据的审计和回溯。存算分离的特性也更加有利于企业做细粒度的成本控制，同时也利于比如在EMR环境下集群的版本升级或管理等。然而存算分离之后会引发数据读取的 IO开销增加，可以通过引入大数据缓存Alluxio以及阿里云的JindoFS 来解决。
上图展示了传统的数仓、数据湖和数据湖仓的区别。可以看到原本数据湖在成本、数据可靠性、监管、安全以及性能上的问题都可以通过数仓的方式得以解决。
二、数据湖格式
数据湖格式是作用在数据湖仓架构之下、在数据真实的存储层之上和元数据管理之下的一层数据存储层。
数据湖的很多特性都由数据湖格式来承载，这也是DeltaLake、Hudi、Iceberg这两年兴起的主要原因和背景。
DeltaLake由Databricks公司提出，近期宣布了将全功能开源，一次性放出很多大家期待的feature。Iceberg由Netflix开源，其设计更开放，在国外的场景上应用更佳。Hudi 由 Uber开源，当前在国内社区比较活跃，其社区风格较为激进，会进行快速功能迭代，因此功能较丰富。Hudi不只是将自己作为数据湖的存储层，而是一个平台，内置了很多湖管理的服务。
数据湖格式的关键实现是：表的元数据不再存放在比如 Hive、MetaStore等服务里，而是和数据一起放在本地的 filesystem。针对于每一次表的变更都会提交新的事务日志，记录对湖格式表的修改、schema的变化和数据文件的变化等。基于以上设计，再结合并发控制协议，能够更好地实现并发控制，拓展 TimeTravel等能力。
经过了几年的发展，三个湖格式的整体功能已经基本一致。
其中Hudi和Iceberg支持的Merge on Read主要解决写数据时遇到的写放大现象。比如想要更新某包含 1 万条数据的文件里的 1000 条数据，需要将 1000 条更新之后的值连同 9000 条没有被修改的数据一起再写回到新的文件。数据湖格式会将前一个版本标记为废弃，将最新版本标记为生效。新版本内的9000条数据原封不动地被保留下来，这就是写放大现象。
而Hudi能够将 1000 条新增的数据通过行存的形式写入增量文件，在查询时合并增量文件和 base 文件，得到全视图最新版本的数据。
DeltaLake 220可以将 delta表作为 CDC 的 source 产生changedata 给下游交付，再做使用。湖管理功能比如合并小文件，用户无需用户再另起任务，能够减少用户的使用和管理成本。Hudi以及 EMR 版本的 DeltaLake都已实现该功能，社区版本目前尚未跟进。
生态上，三个格式也在持续布局和完善。三者都已具备基于 Spark 和 Flink 的读写功能。在查询引擎的支持上略有不同，比如 iceberg在 presto和Jindo上已经支持写的功能。DeltaLake在云厂商的支持以及云catalog 的支持上推动得比较好。
不管哪一种湖格式，都能够实现实时 Sync 、实时入湖和入仓，后续可以进一步做 ETL 处理，比如从 ODS 层到 DWD 到 DWS 层的数据的转换。支持 Spark 等 OLAP 引擎查询，提供近实时的 OLAP 能力。
湖格式场景一：SCD。
SCD会保存随着时间缓慢变化的维度值的变更信息。举个例子，用户和常驻地，用户可能隔几天、几小时会变更所在地，信息随着时间变化。在此类场景下，可能有些数仓无法只提取最新数据，只能提取到某个历史时刻用户在何地。
而SCD根据对新值的处理方式，定义了多种类型Type。Type0指只保留原来的值，将新值丢弃；Type1指只保留最新值，将历史值全部丢弃；Type2指以行存的形式保存所有记录；Type3指增加新的字段来保留前一版本和当前版本。
本示例使用Type2再增加两个字段，用于标识某条记录的生效范围，或者也可以通过布尔值直接表示是否生效。
初始表格如最上方表格所示，需要将原有 ID 标记为无效，最新数据标记为有效；如果某 ID 不存在原表里，则以新增的方式插入。基于以上策略得到的表入上图最下方表格所示。
针对上述示例，传统数仓的实现逻辑比较复杂，很难保证其事务性。而数据湖格式通过语法可在一次事务中完成对数据的删除、修改、新增和插入。
上图右侧为具体语法实现。Using部分的数据通过 when mached（）和 when not mached then（）两部分语法，按照逻辑被写入到 target 表里。when mached语法表示：当匹配到某条记录之后，会对这条记录做删除或修改。when not mached then后面可以加谓词，对数据做插入。
湖格式场景二：G-SCD。
另一类场景，比如只需要知道用户某天内最后在什么地方，即只需知道固定时间段内用户的最新值，可以通过湖格式的TimeTravel解决。我们基于 EMR 胡格式的TimeTravel能力实现了G-SCD解决方案，是基于固定粒度的缓慢变化维。
此前基于 Hive 表的解决方案如下：首先一条实时流通过不断获取增量的数据写入到增量表里。再将T+1的增量表和原始表的 T 分区数据作合并，产生离线表的 T+1分区。通过分区值查询 T+1分区即可获知这一天用户最新的值。该方案存在较重的数据冗余以及资源浪费，因为 T和 T+1 的数据大量重复。
G-SCD的解决方案如下：首先通过 MySQL 将变量信息拖至 Kafka ，消费Kafka数据，通过 Spark Streaming 按 batch 的形式做提交。该方案会要求每一个被提交到 Delta的 commit 只包含业务快照数据，让每一个业务快照和delta版本做关联。比如先提交了一个 T 时刻的版本，又提交了一个T时刻，后面提交了 T+1时刻，则可以通过查询第二个 T时刻的 TimeTravel得到 T时刻的数据。 此外，我们对Spark Streaming做了升级，可按照业务快照做拆分，比如按天级别拆成两天的数据，一天一天地提交，保证提交秩序。
该方案无需添加辅助字段，只需原生的表结构即可解决。同时不存在大量数据冗余，还能够借助于 Spark 和 Delta本身的查询优化提升查询效率。
湖格式场景三：CDC
上图定义了流式的 ETL ，其中表中包含userID、name、city三个字段。通过 user_city 表和 user_name 表做 join ，将 name 字段接入并产生下游的表。再按照 city 表做聚合，得到所在的城常州人口数据。
针对以上需求，此前常用的方案为：首先获取全量的 user_city 表，将它与user_name 表做 join ，以 override 的形式写到 user_name_city 表。然后按照 city 做 group by ，重写后面的 city_population表，每一步都为全量数据。
而CDC方案如下：首先通过 user_name_city表的该次变更提取出其change data，将其与user_name 表做 join 更新到 user_name_city 表。user_name_city 表在这次变更中产生以下变更：用户 1 从北京变为杭州，用户 5 是新增数据，位于武汉。因此，再往下游到 city_population 表里，需要对于北京-1，对杭州和武汉+1，得到其新的常驻人口。
CDC方案需要user 信息、操作，还需要表里的其他字段信息，比如 city 字段等以及获取该记录完整的前值和后值才能对下游做处理。
银行场景下，比如有A 和 B 两个账户， A 账户全天没有任何变化， B 账户全天操作了几百次但最终金额没有变化。此类场景需要追溯每一次变更才有意义。
在EMR内部，Hudi和Delta都是基于 CDLF 的思路实现CDC，充分权衡了读写两端的性能，结合了数据湖格式的事务性特点以及变更的有效映射关系提出的CDLF方案，其核心在于必要时会持久化 CDC数据。类似于数据湖触发器的形式，数据湖可以通过 create trigger 语法来对insert、delete和 update 操作做触发，将前值和旧值写到一张insert表里，后续可直接查询insert表得到其 CDC 数据。
以上方案增加了写开销，但在查询时避免了表 diff ，具有较高的传输效率。而该方案依然存在可优化的点：
首先，尽可能复用已有的数据，避免所有场景、所有写操作都持久化。举个例子，首次表的 insert 。每一条数据都是先插入才进行操作，因此没必要双写一份数CDC数据集，只需下载原有的数据并进行拼接，将其标记为 insert 或前值为空然后返回即可。再比如删除分区场景，整个分区的数据都被删除，因此只需将所有数据加载进来，拼接上删除操作，并置新值为空返回即可。
其次，针对主键场景，仅持久化主键（和操作类型），查询时再提取旧值和新值。
当前，湖格式的使用还存在一些挑战：
第一，学习和使用成本。不同的数据湖格式有不同的实现，在读写场景上也会有所不同，会带来较大的学习成本和后续调优成本。
第二，湖格式统一了批和流，因此必然会有小文件、过期文件清理等问题，管理成本增加。
第三，湖格式依然处于发展阶段，迭代较快，前后版本无法能保证完全兼容。
第四，无法真正地满足OLAP场景的查询性能要求。
第五，引擎之间并不互通，元数据也没有打通，生态集成不够成熟。
第六，业务和客户对于数据湖格式的引入依然持保守观望的态度，更多的使用场景有待挖掘打磨。
三、阿里云EMR+DLF构建湖仓架构
上图为DLF数据湖仓架构图。
最底层由OSS存储作为数据的统一存储层；往上是关键的由 DLF 数据湖构建、开放开源的数据湖格式和数据湖缓存加速JindoFS 三者共同提供的数据湖管理和优化层；再往上是由开源和阿里云自研的产品组成的弹性计算引擎层，提供了基于 BI 和 AI 能力；最上层是由阿里云数据开平台 Dataworks、EMR等提供的开发层，提供了完备的数据开发体系和数据治理平台。
DLF（数据湖构建）是阿里云提出的全托管数据湖服务，提供了湖仓架构下必需的能力。它提供了全托管的统一元数据，借助于该能力可以无缝对接到 EMR 上的各类开源引擎、阿里云的数据湖集成平台以及阿里云的数仓产品等，实现湖仓一体，避免数据孤岛现象。
统一的权限服务能够帮助适配到多个查询引擎和多个产品，为数据的访问和操作提供了安全的保证。标准的入湖模板可以对接到 MySQL Binlog、Kafka等多个 source ，实现一键入湖，可直接保存成湖格式的表，后续再在 EMR 或其他产品上做处理和查询。元数据发现能力可以直接扫描底层的filesystem ，发掘未被管理和记录的数据来消除沉没数据。
数据湖管理能力直接针对于数据湖格式提供的湖管理能力，包括自动化解决小文件以及历史过期文件带来的一系列管理和运营。
DLF会监听湖格式每一次提交的 commit ，感知每一次变化，实时分析各种维度的指标，包括文件数、平均文件大小、有效/无效文件占比等。
EMR的数据湖格式提供了丰富的SQL语法，包括 Partition相关、DPO、Optimize、TimeTravel等，支持Savepoint、RollBack等，拓展了Spack SQL支持Streaming SQL语法。EMR解决了湖格式存在的问题，保证了湖格式和普通的 Hive 表的功能和场景性能一致。即使在存在schema 变更的情况下也能够保证正确、实时地同步到相关元数据服务。三种湖格式都可对接到 DLF ，支持阿里云其他产品的查询。
我们也在积极拥抱社区，向社区贡献了很多关键 feature，比如Spark3.0 的 Merge Into语法、Delta支持Hive查询的delta-connectors以及Hudi的Spark SQL支持和Hudi CDC。
Q&A
Q：湖格式产品支持 CDC，也是通过 Binlog 实现的吗？
A：湖格式产品支持CDC，是将自己的表作为 CDC 的 source 提供服务。主要用于构建完整的一套增量式场景。此前通过 SQL Binlog 只是将数据库里的changedata做入湖而已，并不能继续支持下游的场景。
Q：请问 Hudi 现在支持使用 Java reader读取数据吗？
A：支持，可以通过 Java 的方式读写数据，但性能上不如Flink 或 Spark等大数据引擎。
