本文是字节跳动数据平台开发套件团队在 Flink Forward Asia 2021: Flink Forward 峰会上的演讲，着重分享了字节跳动数据湖技术上的选型思考和摸索实际。
文 | Gary Li 字节跳动数据平台开发套件团队高级研发工程师，数据湖开源我的项目 Apache Hudi PMC Member
随着 Flink 社区的一直倒退，越来越多的公司将 Flink 作为首选的大数据计算引擎。字节跳动也在继续摸索 Flink，作为泛滥 Flink 用户中的一员，对于 Flink 的投入也是逐年减少。
字节跳动数据集成的现状
在 2018 年，咱们基于 Flink 结构了异构数据源之间批式同步通道，次要用于将在线数据库导入到离线数仓，和不同数据源之间的批式传输。
在 2020 年，咱们基于 Flink 结构了 MQ-Hive 的实时数据集成通道，次要用于将音讯队列中的数据实时写入到 Hive 和 HDFS，在计算引擎上做到了流批对立。
到了 2021 年，咱们基于 Flink 结构了实时数据湖集成通道，从而实现了湖仓一体的数据集成系统的构建。
字节跳动数据集成系统目前反对了几十条不同的数据传输管道，涵盖了线上数据库，例如 Mysql Oracle 和 MangoDB；音讯队列，例如 Kafka RocketMQ；大数据生态系统的各种组件，例如 HDFS、HIVE 和 ClickHouse。
在字节跳动外部，数据集成系统服务了简直所有的业务线，包含抖音、今日头条等大家耳熟能详的利用。
整个零碎次要分成 3 种模式——批式集成、流式集成和增量集成。
- 批式集成模式基于 Flink Batch 模式打造，将数据以批的模式在不同零碎中传输，目前反对了 20 多种不同数据源类型。
- 流式集成模式次要是从 MQ 将数据导入到 Hive 和 HDFS，工作的稳定性和实时性都受到了用户宽泛的认可。
- 增量模式即 CDC 模式，用于反对通过数据库变更日志 Binlog，将数据变更同步到内部组件的数据库。
这种模式目前反对 5 种数据源，尽管数据源不多，然而工作数量十分宏大，其中蕴含了很多外围链路，例如各个业务线的计费、结算等，对数据准确性要求十分高。**
在 CDC 链路的整体链路比拟长。首先，首次导入为批式导入，咱们通过 Flink Batch 模式直连 Mysql 库拉取全量数据写入到 Hive，增量 Binlog 数据通过流式工作导入到 HDFS。
因为 Hive 不反对更新操作，咱们仍旧应用了一条基于 Spark 的批处理链路，通过 T-1 增量合并的形式，将前一天的 Hive 表和新增的 Binlog 进行合并从而产出当天的 Hive 表。
随着业务的疾速倒退，这条链路裸露进去的问题也越来越多。
- 首先，这条基于 Spark 的离线链路资源耗费重大，每次产出新数据都会波及到一次全量数据 Shuffle 以及一份全量数据落盘，两头所耗费的贮存以及计算资源都比较严重。
- 同时，随着字节跳动业务的疾速倒退，近实时剖析的需要也越来越多。
- 最初，整条链路流程太长，波及到 Spark 和 Flink 两个计算引擎，以及 3 个不同的工作类型，用户应用老本和学习老本都比拟高，并且带来了不小的运维老本。
为了解决这些问题，咱们心愿对增量模式做一次彻底的架构降级，将增量模式合并到流式集成中，从而能够解脱对 Spark 的依赖，在计算引擎层面做到对立。
革新实现后，基于 Flink 的数据集成引擎就能同时反对批式、流式和增量模式，简直能够笼罩所有的数据集成场景。
同时，在增量模式上，提供和流式通道相当的数据提早，赋予用户近实时剖析能力。在达到这些指标的同时，还能够进一步升高计算成本、提高效率。
通过一番摸索，咱们关注到了正在衰亡的数据湖技术。
对于数据湖技术选型的思考
咱们的眼光集中在了 Apache 软件基金会旗下的两款开源数据湖框架 Iceberg 和 Hudi 中。
Iceberg 和 Hudi 两款数据湖框架都十分优良。但两个我的项目被创立的目标是为了解决不同的问题，所以在性能上的侧重点也有所不同：
- Iceberg：外围形象对接新的计算引擎的老本比拟低，并且提供先进的查问优化性能和齐全的 schema 变更。
- Hudi：更重视于高效率的 Upsert 和近实时更新，提供了 Merge On Read 文件格式，以及便于搭建增量 ETL 管道的增量查问性能。
一番比照下来，两个框架各有千秋，并且离咱们设想中的数据湖最终状态都有肯定间隔，于是咱们的外围问题便集中在了以下两个问题：
哪个框架能够更好的反对咱们 CDC 数据处理的外围诉求？
哪个框架能够更疾速补齐另一个框架的性能，从而成长为一个通用并且成熟的数据湖框架？
通过屡次的外部探讨，咱们认为：Hudi 在解决 CDC 数据上更为成熟，并且社区迭代速度十分快，特地是最近一年补齐了很多重要的性能，与 Flink 的集成也愈发成熟，最终咱们抉择了 Hudi 作为咱们的数据湖底座。
01 – 索引零碎
咱们抉择 Hudi，最为看重的就是 Hudi 的索引零碎。
这张图是一个有索引和没有索引的比照。
在 CDC 数据写入的过程中，为了让新增的 Update 数据作用在底表上，咱们须要明确晓得这条数据是否呈现过、呈现在哪里，从而把数据写到正确的中央。在合并的时候，咱们就能够只合并单个文件，而不须要去管全局数据。
如果没有索引，合并的操作只能通过合并全局数据，带来的就是全局的 shuffle。
在图中的例子中，没有索引的合并开销是有索引的两倍，并且如果随着底表数据量的增大，这个性能差距会呈指数型回升。
所以，在字节跳动的业务数据量级下，索引带来的性能收益是十分微小的。
Hudi 提供了多种索引来适配不同的场景，每种索引都有不同的优缺点，索引的抉择须要依据具体的数据分布来进行取舍，从而达到写入和查问的最优解。
上面举两个不同场景的例子。
1、日志数据去重场景
在日志数据去重的场景中，数据通常会有一个 create_time 的工夫戳，底表的散布也是依照这个工夫戳进行分区，最近几小时或者几天的数据会有比拟频繁的更新，然而更老的数据则不会有太多的变动。
冷热分区的场景就比拟适宜布隆索引、带 TTL 的 State 索引和哈希索引。
2、CDC 场景
第二个例子是一个数据库导出的例子，也就是 CDC 场景。这个场景更新数据会随机散布，没有什么法则可言，并且底表的数据量会比拟大，新增的数据量通常相比底表会比拟小。
在这种场景下，咱们能够选用哈希索引、State 索引和 Hbase 索引来做到高效率的全局索引。
这两个例子阐明了不同场景下，索引的抉择也会决定了整个表读写性能。Hudi 提供多种开箱即用的索引，曾经笼罩了绝大部分场景，用户应用老本非常低。
02 – Merge On Read 表格局
除了索引零碎之外，Hudi 的 Merge On Read 表格局也是一个咱们看重的外围性能之一。这种表格局让实时写入、近实时查问成为了可能。
在大数据体系的建设中，写入引擎和查问引擎存在着人造的抵触：
写入引擎更偏向于写小文件，以行存的数据格式写入，尽可能防止在写入过程中有过多的计算包袱，最好是来一条写一条。
查问引擎则更偏向于读大文件，以列存的文件格式贮存数据，比如说 parquet 和 orc，数据以某种规定严格散布，比方依据某个常用字段进行排序，从而做到能够在查问的时候，跳过扫描无用的数据，来缩小计算开销。
为了在这种人造的抵触下找到最佳的取舍，Hudi 反对了 Merge On Read 的文件格式。
MOR 格局中蕴含两种文件：一种是基于行存 Avro 格局的 log 文件，一种是基于列存格局的 base 文件，包含 Parquet 或者 ORC。
log 文件通常体积较小，蕴含了新增的更新数据。base 文件体积较大，蕴含了所有的历史数据。
- 写入引擎能够低提早的将更新的数据写入到 log 文件中。
- 查问引擎在读的时候将 log 文件与 base 文件进行合并，从而能够读到最新的视图；compaction 工作定时触发合并 base 文件和 log 文件，防止 log 文件继续收缩。在这个机制下，Merge On Read 文件格式做到了实时写入和近实时查问。
03 – 增量计算
索引零碎和 Merge On Read 格局给实时数据湖打下了十分松软的根底，增量计算则是这个根底之上的 Hudi 的又一个亮眼性能：
增量计算赋予了 Hudi 相似于音讯队列的能力。用户能够通过相似于 offset 的工夫戳，在 Hudi 的工夫线上拉取一段时间内的新增数据。
在一些数据提早容忍度在分钟级别的场景中，基于 Hudi 能够对立 Lambda 架构，同时服务于实时场景和离线场景，在贮存上做到流批一体。
字节跳动外部场景实际思考
在抉择了基于 Hudi 的数据湖框架后，咱们基于字节跳动外部的场景，打造定制化落地计划。咱们的指标是通过 Hudi 来反对所有带 Update 的数据链路：
须要高效率且低成本的 Upsert
反对高吞吐
端到端的数据可见性管制在 5-10 分钟以内
指标明确后，咱们开始了对 Hudi Flink Writer 进行了测试。这个图是 Hudi on Flink Writer 的架构：一条新的数据进来之后，首先会通过一个索引层，从而找到它须要去的中央。
- State 索引中保留了所有主键和文件 ID 的一一映射关系，对于 Update 数据，会找到其所存在的文件 ID，对于 Insert 数据，索引层会给他指定一个新的文件 ID，或者是历史文件中的小文件，让其填充到小文件中，从而防止小文件问题。
- 通过索引层之后，每条数据都会带有一个文件 ID，Flink 会依据文件 ID 进行一次 shuffle，将雷同文件 ID 的数据导入到同一个子工作中，同时能够防止多个工作写入同一个文件的问题。
- 写入子工作中有一个内存缓冲区，用于贮存以后批次的所有数据，当 Checkpoint 触发时，子工作缓冲区的数据会被传入 Hudi Client 中，Client 会去执行一些微批模式的计算操作，比方 Insert/Upsert/Insert overwrite 等，每种操作的计算逻辑不同，比如说 Insert 操作，会生成一个新的文件，Upsert 操作可能会和历史文件做一次合并。
- 待计算实现后，将解决好的数据写入到 HDFS 中，并同时收集元数据。
- Compaction 工作为流工作的一部分，会定时的去轮训 Hudi 的工夫线，查看是否有 Compaction 打算存在，如果有 Compaction 打算，会通过额定的 Compaction 算子来执行。
在测试过程中，咱们遇到了以下几个问题：
- 在数据量比拟大的场景下，所有的主键和文件 ID 的映射关系都会存在 State 中，State 的体积收缩的十分快，带来了额定的贮存开销，并且有时会造成 Checkpoint 超时的问题。
- 第二个问题是，因为 Checkpoint 期间，Hudi Client 操作比拟重，比如说和底层的 base 文件进行合并，这种操作波及到了历史文件的读取，去重，以及写入新的文件，如果遇到 HDFS 的抖动，很容易呈现 Checkpoint 超时的问题
- 第三个问题是，Compaction 工作作为流式工作的一部分，工作启动后资源就不可调节，如果须要调节，只能重启整个工作，开销比拟大，如果不能灵便调节 Compaction 工作，就可能会呈现 Compaction 算子空跑导致资源节约，或者资源有余导致工作失败的状况。
为了解决这些问题，咱们开始针对咱们的场景进行了定制化的优化。
字节跳动的定制化优化技术计划
01 – 索引层
索引的目标就是找到以后这条数据所在的文件地点，存在 State 中的话每条数据都波及到一次 State 的读和写，在数据量大的场景下，所带来的计算和贮存开销都是比拟大的。
字节跳动外部开发了一种基于哈希的索引，能够通过间接对主键的哈希操作来找到文件所在的地位，这种形式在非分区表下能够做到全局索引，绕过了对 State 的依赖，革新过后，索引层变成了一层简略的哈希操作。
02 – 写入层
晚期的 Hudi 写入和 Spark 强绑定，在 2020 年底，Hudi 社区对底层的 Hudi Client 进行了拆分，并且反对了 Flink 引擎，这种革新形式是将 Spark RDD 的操作变成了一个 List 的操作，所以底层还是一个批式操作，对于 Flink 来说，每一次 Checkpoint 期间所须要做的计算逻辑是相似于 Spark RDD 的，相当于是执行了一次批式的操作，计算包袱是比拟大的。
写入层的具体流程是：一条数据通过索引层后，来到了写入层，数据首先会在 Flink 的内存缓冲区积攒，同时通过内存监控来防止内存超出限度导致工作失败，到了 Checkpoint 的时候，数据会被导入到 Hudi Client，而后 Hudi Client 会通过 Insert，Append，Merge 等操作计算最终的写入数据，计算实现后将新的文件写入到 HDFS 并同时回收元数据。
咱们的外围指标在于如何让这种微批的写入模式更加的流式化，从而升高 Checkpoint 期间的计算累赘。
在表构造上，咱们抉择了与流式写入更加匹配的 Merge on Read 格局，写入的算子只负责对于 log 文件的追加写入，不做任何别的额定的操作，例如和 base 文件进行合并。
在内存上，咱们将第一层 Flink 的内存缓冲区去掉，间接把内存缓冲区建设在了 hudi client 中，在数据写入的同时进行内存监控防止内存超出限度的状况，咱们将写入 hdfs 的操作和 Checkpoint 进行理解耦，工作运行过程中，每一小批数据就会写入 HDFS 一次，因为 HDFS 反对追加写操作，这种模式也不会带来小文件的问题，从而将 Checkpoint 尽可能的轻量化，防止 HDFS 抖动和计算量过大带来的 Checkpoint 超时的问题。
03- Compaction 层
Compaction 工作实质上是一个批工作，所以须要和流式写入进行拆分，目前 Hudi on Flink 反对了异步执行 Compaction 的操作，咱们的线上工作全副应用了这种模式。
在这种模式下，流式工作能够专一于写入，晋升吞吐能力和进步写入的稳定性，批式的 Compaction 工作能够流式工作解耦，弹性伸缩高效的利用计算资源，专一于资源利用率和节约老本。
在这一系列的优化过后，咱们在一个 2 百万 rps 的 Kafka 数据源上进行了测试，应用了 200 个并发导入到 Hudi。和之前相比，Checkpoint 耗时从 3-5 分钟升高到了 1 分钟以内，HDFS 抖动带来的工作失败率也大幅度降落因为 Checkpoint 耗时升高，理论用于数据处理的工夫变得更多了，数据吞吐量翻了一倍，同时 State 的存储开销也降到了最低。
这是最终的 CDC 数据导入流程图
首先，不同的数据库会将 Binlog 发送到音讯队列中，Flink 工作会将所有数据转换成 HoodieRecord 格局，而后通过哈希索引找到对应的文件 ID，通过一层对文件 ID 的 shuffle 后，数据达到了写入层，写入算子以追加写的模式将数据频繁的写入到 HDFS 中，Checkpoint 触发后，Flink 会将所有的元数据收集到一起，并写入到 hudi 的元数据系统中，这里就标记了一个 Commit 提交实现，一个新的 Commit 会随之开始。
用户能够通过 Flink Spark Presto 等查问引擎，近实时的查问曾经提交实现的数据。
数据湖平台侧托管的 Compaction 服务会定时提交 Flink Batch 模式的 Compaction 工作，对 Hudi 表进行压缩操作，这个过程对用户无感知并且不影响写入工作。
咱们这一整套解决方案也会奉献给社区，感兴趣的同学能够关注 Hudi 社区最新的停顿
流式数据湖集成框架的典型落地场景
流式数据湖集成框架革新实现后，咱们找到了一些典型的落地场景：
利用最广泛的就是将线上数据库导入到离线数仓进行剖析的场景，和之前的 Spark 离线链路相比：端到端的数据提早从一个小时以上升高到了 5-10 分钟，用户能够进行近实时的数据分析操作。
在资源利用率方面，咱们模仿了一个 Mysql 导入离线数仓进行剖析的场景，将 Flink 流式导入 Hudi 和 Spark 离线合并的计划进行了比照，在用户小时级查问的场景下，端到端的计算资源大概节约了 70% 左右。
在字节跳动 EB 级数据量的数仓场景下，这种资源利用率的晋升所带来的收益是十分微小的。
对于基于音讯队列和 Flink 构建实时数仓的用户来说，他们能够把不同数仓层级的实时数据导入到 Hudi，这类数据 update 的状况很多，所以相较于 Hive，Hudi 能够提供高效率且低成本的 Upsert 操作，从而用户能够对于全量数据进行近实时查问，防止了一次去重的操作。
这是一个 Flink 双流 Join 的场景，很多 Flink 的用户会应用双流 Join 来进行实时的字段拼接，在应用这个性能的时候，用户通常会开一个工夫窗口，而后将这个工夫窗口中来自不同数据源的数据拼接起来，这个字段拼接性能也能够在 Hudi 的层面实现。
咱们正在摸索一个性能，在 Flink 中只将不同 Topic 的数据 Union 在一起，而后通过 Hudi 的索引机制，将雷同主键的数据都写入到同一个文件当中，而后通过 Compaction 的操作，将数据进行拼接。
这种形式的长处在于，咱们能够通过 Hudi 的索引机制来进行全局字段拼接，不会受到一个窗口的限度。
整个拼接逻辑通过 HoodiePayload 实现，用户能够简略的继承 HoodiePayload，而后来开发本人的自定义的拼接逻辑，拼接的机会能够是 Compaction 工作，也能够是 Merge on Read 近实时查问，用户能够依据需要场景，灵便的应用计算资源。然而相比 Flink 双流 Join，这种模式会有一个毛病，就是实时性和易用性上要差一些。
结语
在这一系列的工作过后，咱们对数据湖的将来满怀期待，同时也设立的明确的指标。
首先，咱们心愿将 Hudi 作为所有 CDC 数据源的底层存储，齐全替换掉基于 Spark 的离线合并计划，通过数据集成引擎流式导入，将近实时离线剖析的能力带给所有的在线数据库。
接着，增量 ETL 场景也是一个重要的落地场景，对于数据提早容忍度在分钟级的场景，Hudi 能够作为对立存储同时服务于实时链路和离线链路，从而将传统的数仓 Lambda 架构降级到真正意义上的流批一体。
最初，咱们心愿建设一个智能数据湖平台，这个平台会托管所有数据湖的运维治理，达到自我治理的一个状态，用户则不须要再为运维而懊恼。
同时，咱们心愿提供自动化调优的性能，基于数据的散布找到最佳的配置参数，例如之前提到的不同索引之间的性能取舍问题，咱们心愿通过算法来找到最佳的配置，从而进步资源利用率，并升高用户的应用门槛。
极佳的用户体验也是咱们的谋求之一，咱们心愿在平台侧做到一键入湖入仓，大大降低用户的开发成本。
数据湖集成技术也曾经通过火山引擎大数据研发治理套件 DataLeap 对外开放。
欢送关注字节跳动数据平台同名公众号