阿里云ADB基于Hudi构建Lakehouse的实践

　　导读： 大家好，我是来自阿里云数据库的李少锋，现在主要专注于 ADB Hudi & Spark 的研发以及产品化，今天非常高兴能够借这个机会和大家分享下阿里云 ADB 基于 Apache Hudi 构建 Lakehouse 的应用与实践。
　　接下来我将分为 3 个部分给大家介绍今天的议题，首先我会介绍经过将近一年打磨推出的ADB 湖仓版的架构以及关键优势，接着会介绍在支持客户构建 Lakehouse 时，我们是如何克服基于 Hudi 构建千亿数据入湖的挑战；最后将介绍基于 ADB 构建 Lakehouse 的实践。
　　分享嘉宾 李少锋 阿里云 数据库技术专家
　　出品社区 DataFun
　　01
　　ADB 湖仓版机构与关键优势
　　首先先来介绍下 ADB 湖仓版架构及其关键优势
　　一体版本，我们称为 ADB 湖仓版。湖仓版在数据全链路的「采存算管用」5 大方面都进行了全面升级。
　　在「采集」方面 ，我们推出了数据管道 APS 功能，可以一键低成本接入数据库、日志、大数据中的数据。
　　在「存储」方面 ，我们除了新增 Hudi 格式的外表能力，也对内部存储进行了升级改造。通过只存一份数据，同时满足离线、在线 2 类场景。
　　在「计算」方面 ，我们对自研的 XIHE BSP SQL 引擎进行容错性、运维能力等方面的提升，同时引入开源 Spark 引擎满足更复杂的离线处理场景和机器学习场景。
　　在「管理」方面 ，我们推出了统一的元数据管理服务，统一湖仓元数据及权限，让湖仓数据的流通更顺畅
　　在「应用」方面 ，除了通过SQL方式的BI分析应用外，还支持基于 Spark 的 AI 应用。
　　我们希望通过资源一体化和体验一体化，2 个一体化能力，帮助客户实现降本增效。资源一体化是指一份数据、极致弹性和融合引擎。体验一体化是指统一的计费单位、数据管道、数据管理和数据访问。
　　在 ADB 湖仓版中，首先将一份全量数据存在低成本高吞吐存储介质上，低成本离线处理场景直接读写低成本存储介质，降低数据存储和数据 IO 成本，保证高吞吐。其次将实时数据存在单独的存储 IO 节点（EIU）上，保证「行级」的数据实时性，同时对全量数据构建索引，并通过 Cache 能力对数据进行加速，满足百 ms 级高性能在线分析场景。因此，数据湖中的数据可以在数仓中加速访问；而数仓中的数据，也可以利用离线资源进行计算。
　　极致弹性通过弹得好、弹得起、弹得快三个特点，贴合业务负载，保证查询性能，降低资源成本。
　　弹得好 是指提供了适合在线分析业务的分时弹性和适合离线处理、机器学习的按需弹性，完美贴合业务负载，满足业务需求。
　　弹得起 是指基于神龙 + ECS + ECI构建了三级资源库存供给能力，保障弹性资源交付率超过 95%。
　　弹得快 是指资源池化后通过缓存加速等技术，弹性启动效率可以达到 10s 内。
　　支撑离线、在线两类场景的背后，除了刚才提到的一份数据。还有自研的XIHE融合计算引擎，同时提供 MPP 模式和 BSP 模式，并提供自动切换能力。
　　自动切换能力是指当查询使用 MPP 模式无法在一定耗时内完成时，系统会自动切换为 BSP模式进行执行。未来，我们还将在一个查询中，根据算子特点部分算子采用 MPP 模式，部分算子采用 BSP 模式，兼顾高性能和高吞吐，提供更智能的查询体验。
　　同时融合引擎为提供资源利用率提供了可能，通常离线处理发生在晚上，在线分析发生在白天，一份资源可以同时支持 2 类场景，通过错峰提升资源利用率。
　　最后再介绍一下统一数据管道APS。数据快速接入，是数据分析的第一步，也是最容易流失客户的一步。但我们发现数据接入面临体验差、成本高、门槛高等痛点。
　　所以，我们决定跟其它接入工具做好深度优化的同时，面向中小客户，构建一个统一数据管道 APS，底层基于 Flink 实时引擎，提供易用性好、低延时、低成本的数据接入体验。
　　对于湖仓中的表元数据，ADB 做了统一元数据服务，湖仓中的元数据/权限可互通，可通过不同引擎自由访问湖仓数据；而对于湖仓数据，为了屏蔽底层数据存储格式的差异，便于第三方引擎集成，ADB 提供了面向内存列存格式 Arrow，满足对吞吐的要求，对于内部存储，已经通过 Arrow 格式完成 Spark 对接，提供高吞吐能力。
　　自研是打造技术深度的基础，但同时我们积极拥抱开源，满足已经生长在开源生态上的客户可以更平滑地使用湖仓版。外表类型，在 Parquet/ORC/JSON/CSV 等 Append 类型数据格式的基础上，为支持在对象存储上低成本 Lakehouse，支持了Apache Hudi，提供近实时的更新、删除能力，满足客户对低成本的诉求。
　　--
　　02
　　基于 Hudi 构建千亿数据入湖的挑战
　　以上就是 ADB 湖仓版的架构与关键优势，接着介绍基于 Hudi 构建千亿数据入湖的挑战以及如何我们是如何克服这些挑战的。
　　首先我们看下典型的业务场景，业务源端数据通过数据采集进入阿里云 SLS，然后通过 APS数据管道服务进入ADB 湖仓版，基于 ADB 湖仓版之上构建日志查询、日志导出、日志分析等功能。
　　该业务场景有如下典型的特点：
　　1.高吞吐，单链路最高 4GB/s 吞吐，日增数据量 350TB，总存储量超 20PB；
　　2.数据倾斜/热点严重：源端数据倾斜非常严重，从百万到几十条数据不等；
　　3.扫描量大：日志查询的扫描量 50GB ~ 500GB 不等，查询并发在 100+；
　　如果使用数仓的话会面临成本高的问题，另外是数仓缺少热点打散能力，只能不断加资源，瓶颈明显；最后是数仓系统资源是固化的，没有动态弹性能力，或者弹性能力较弱，承载不同客户查询需求时，容易互相干扰，尤其是大客户的数据扫描。
　　先来看下 日志数据入湖的技术架构 ，数据从 SLS 读取，然后通过 Flink 写入 Hudi，整个架构非常简单，其中对于 SLS 和 Flink 的状态存储说明如下：
　　1.SLS 多 Shard 存储，由 Flink 的多个 Source 算子并行消费
　　2.消费后通过 Sink 算子调用 Hudi Worker/Writer 写出到 Hudi（实际链路还存在 Repartition，热点打散等逻辑）
　　3.Flink Checkpoint 后端存储以及 Hudi 数据存储在OSS
　　接着来看下  Flink 写入 Hudi 的主流程 ，首先明确 Flink 写 Hudi 时会存在两种角色，Coordinator 负责处理元数据，如对 Hudi Instant 的相关操作，Worker/Writer 负责处理数据，如写入数据为 parquet 格式，写入步骤如下
　　1.Coordinator 开启一个 Hudi Instant
　　2.Filnk Sink 发送数据给 Hudi Worker 写出
　　3.触发 Flink Checkpoint 时，则通过 Sink 算子通知 Worker Flush 数据，同时持久化operator-state
　　4.当 Flink 完成 Checkpoint 持久化时，通知 Coordinator 提交该 Instant，Coordinator 完成最终提交，写 commit 文件，数据对外可见
　　5.提交后会立即开启一个新的 Instant，继续上述循环
　　如果把 Flink 写 Hudi 保证端到端一致性分成两部分，一部分是 Flink 框架内部的，另外一个部分是与 Hudi 交互的部分。
　　1.其中步骤 1 到 3 之间是 Flink Checkpoint 逻辑，如果异常在这些步骤上发生，则认为 Checkpoint失败，触发 Job 重启，从上一次 Checkpoint 恢复，相当于两阶段提交的 Precommit 阶段失败，事务回滚，如果有 Hudi 的 inflight commit，等待 Hudi Rollback 即可，无数据不一致问题
　　2.当 3 到 4 之间发生异常，则出现 Flink 和 Hudi 状态不一致。此时 Flink 认为 Checkpoint 已结束，而 Hudi 实际尚未提交。如果对此情况不做处理，则发生了数据丢失，因为Flink Checkpoint 完毕后，SLS 位点已经前移，而这部分数据在 Hudi 上并未完成提交，因此容错的重点是如何处理此阶段引起的数据一致性问题
　　3.我们拿一个例子举例分析在步骤 3 和 4 之前发生异常时，如果保证数据一致性
　　4.否则认为上一次 Instant 执行失败，等待 Rollback 即可，脏数据对用户不可见
　　我们举例分析下在步骤 3 和 4 之间发生异常时，是如何保证数据一致性的，可以看到对于1615 的 commit 在 Flink 完成 Checkpoint 时会将其 instant 信息持久化至 Flink 后端存储。
　　从 checkpoint 恢复时有如下步骤：
　　1.Checkpoint 时 Sink 算子 Flush 数据及持久化 Instant 的 state；
　　2.Worker 请求处于 pending 的 Instant，与从 state 恢复的 Instant 做对比并汇报给 Coordinator；
　　3.Coordinator 从 Instant Timeline 中获取最新的 Instant 信息，并接收所有 Worker 的汇报；
　　4.如果 Worker 汇报 Instant 相同，并且不在 Timeline 中已完成的 Instant 中，则表示该 Instant 尚未提交，触发 Recommit。
　　经过上述步骤可以保证在 Flink 完成 Checkpoint 时，但对于 Hudi Commit 失败时的场景会进行 recommit，从而保证数据的一致性。
　　接着介绍 我们在处理 4GB/s 的高吞吐时面临的挑战 ，一个非常大的挑战就是热点数据处理，我们统计了 5 分钟内各 Task 处理数据的大小，发现各 Task 处理数据从 200W 条到几十条不等，热点问题明显。
　　而在写入链路中会根据分区字段做 shuffle，同一个分区由一个 Task 写入,对于上述存在的热点问题会导致部分TM上的分区写入非常慢，导致数据延迟/作业挂掉。
　　面对写入热点问题，我们开发了热点打散功能，通过配置指定规则打散热点数据，同时可以根据数据流量自动更新热点打散规则，确保系统的健壮性，可以看到经过热点打散后个 TM 处理的数据量/CPU占用/内存占用基本相同并且比较平稳，作业稳定性也得到了提升。
　　另外一个挑战是 OOM，其实和热点打散也有很大关系，我们发现作业运行时会出现OOM，导致作业挂掉，数据延迟上涨，因此 我们对堆外/堆内内存的使用做了比较细致的梳理， 使用内存的部分主要集中在：
　　1.写 Parquet 文件占用堆外内存
　　2.OSS Flush 占用堆外内存
　　3.单 TM 的 Slot 数、写并发都影响内存占用，如每个写并发处理 30-50 Handle，TM 16 并发，8M row group size 最多占用 6400 M 内存
　　4.堆内内存负载过高导致频繁Full GC
　　我们针对上述 分内存使用做了优化 ，如：
　　1.row group size 配置为 4M，减少堆外内存占用，同时将堆外内存调大
　　2.close 时及时释放 compressor 占用的内存，这部分对 parquet 源码做了改造
　　3.透出堆外内存指标，增加堆外内存监控，这部分也是对 parquet 源码做了改造
　　4.源端 source 算子与 Shard 分配更均衡，以此保证各 TM 消费的 shard 数基本均等
　　最后一个比较大的挑战就是 OSS 限流，云对象存储（如OSS）对 List 操作不友好，list objects 对 OSS 服务器压力较大，如在我们场景下，1500 写并发，会产生 1W QPS list object，OSS 侧目前限流 1K QPS，限流明显，限流会导致作业处理变慢，延迟变高，为解决该问题， 我们也梳理了写入链路对 OSS 的请求，在元数据链路对 OSS 的请求如下：
　　1.Timeline 构建需要 list .hoodie 目录
　　2.Flink CkpMetaData 基于 OSS 传递给 Worker
　　3.Hadoop-OSS SDK create/exists/mkdir 函数依赖 getStatus 接口，getStatus 接口现有实现导致大量 list 请求，其中 getStatus 接口对于不存在的文件，会额外进行一次 list objects 请求确认 Path 是不是目录，对 Marker File、partitionMetadata、数据文件都会产生大量的 list objects 请求
　　在数据链路对 OSS 请求如下：先临时写到本地磁盘，再上传至 OSS，导致本地磁盘写满。
　　针对上述对 OSS 的请求，我们做了如下优化，在元数据侧：
　　1.Timeline Based CkpMetaData，将TM请求打到 JM，避免大量 TM 扫描 OSS 文件
　　2.Hadoop-OSS SDK，透出直接创建文件的接口，不进行目录检查
　　3.PartitionMetaData 缓存处理，在内存中对每个分区的元数据文件做了缓存处理，尽量减少与 OSS 的交互
　　4.Create Marker File 异步处理，异步化处理不阻塞对 Handle 的创建，减少创建 Handle 的成本
　　5.开启 Timeline Based Marker File，这个是社区已经有的能力，直接开启即可
　　这里额外补充下可能有小伙伴比较好奇为什么开启 hudi metadata table 来解决云对象存储的限流问题，我们内部做过测试，发现开启 metadata table 时，写入会越来越慢，无法满足高吞吐的场景。
　　以上就是我们在处理日志数据入湖时面临的典型挑战以及我们如何克服这些挑战，接着讲讲我们在处理数据入湖时为满足业务要求做的关键特性开发。
　　首先是支持并发写，业务侧要求链路有补数据能力，补数据场景涉及多 Flink Client 写不同分区，实时写链路，补数据链路，Table Service 链路并发操作表数据/元数据，这要求：
　　1.表数据不错乱
　　2.补数据/TableService 链路不影响实时写链路
　　因此 我们对 Hudi 内核侧做了部分修改来支持并发写场景：
　　1.CkpMetadata 唯一标识，保证不同作业使用不同 ckp meta
　　2.ViewStorage 唯一标识，保证不同作业 Timeline Server 隔离
　　3.Lazy Table Service，保证并行作业不互相 rollback commit，避免数据错乱
　　4.Instant 生成重试策略，保证 Timeline Instant 的唯一性，避免数据错乱
　　5.独立 Table Service 处理，使用单独的作业运行 Table Service，与实时写链路完全隔离
　　另外一个关键特性是出于成本考虑，业务侧要求 Hudi 中数据不能无限地保存，需要按照用户设定的策略保留指定时间的数据，这要求：
　　1.Hudi 提供分区级别按照数据量，分区数和过期时间等不同维度进行生命周期管理的能力
　　2.Hudi 支持并发设置生命周期管理策略，因为面向多租户会涉及并发更新管理策略
　　针对业务对生命周期管理的需求，我们开发 Hudi 的生命周期管理功能，具体实现如下：
　　1.对于生命周期管理使用，首先通过 call command 添加生命周期管理策略，并进行持久化，为支持并发更新，我们参考 Hudi MOR 表中 Base 文件和 Log 文件的设计，并发更新会写入不同的 Log 文件；
　　2.对于生命周期管理的执行，在每一次 commit 结束后进行统计信息增量采集并更新至统计信息文件，然后按照分区策略进行过期分区的处理，对于过期分区会生成一个 replace commit，等待后续被 clean 即可，同时会合并前面的策略 Base 文件和 Log 文件，生成新的 Base 文件以及更新统计信息；
　　我们也提供了按照分区数、数据量、过期时间三种不同策略来管理 Hudi 表中的分区的生命周期，很好的满足业务侧的需求。
　　最后一个比较大的关键特性是独立 TableService，业务侧要求保证实时写链路稳定，同时希望提高入湖数据的查询性能，这要求：
　　1.在不影响主链路同步性能情况下，清理 Commits/文件版本，保证表状态大小可控
　　2.为提高查询性能，提供异步 Clustering 能力，合并小文件，减少扫描量，提高查询性能
　　基于上述诉求我们开发了基于 ADB 湖仓版的独立 Table Service 服务，在入湖链路写入完成后会进行一次调度，然后将请求写入调度组件，供调度组件后续拉起弹性的 Flink/Spark TableService 作业，可以做到对实时写入链路无影响。
　　对于表状态管理以及数据布局优化均是采用的独立 TableService 作业执行，保证表的状态大小可控，同时通过异步 Clustering 操作，端到端查询性能提升 40% 以上。
　　在对日志入湖链路进行了深入打磨后，我们可以保证最高 4GB/s 的数据写入，延迟在 5min内，满足业务需求。
　　同时也建设了指标监控大盘与异常链路告警功能，便于快速定位问题以及出现问题后快速响应。
　　以上介绍便是我们是如何基于Hudi构建千亿数据入湖以及如何克服入湖挑战的。
　　--
　　03
　　基于 ADB 构建 Lakehouse 的实践
　　最后介绍下基于 ADB 构建 Lakehouse 的实践。
　　前面也提到 ADB 湖仓版拥抱开源技术，ADB 集成了流式处理引擎 Flink，并在此基础上推出了  APS 数据管道服务，APS 具备如下优势：
　　1.低成本，低延迟：作业级别弹性资源，按量付费；按流量自由设定作业资源；充分享受 Flink 流式处理性能红利
　　2.多数据源快速集成：得益于 Flink 成熟的 Connectors 机制，可以方便对接如 SLS、Kafka 等数据源，同时可以保证数据入湖的精确一致性
　　3.低使用门槛：支持白屏化操作快速构建 Lakehouse，基于统一元数据服务，Lakehouse 数据可通过 Spark/ADB 引擎无缝访问
　　而为了满足客户对于批处理以及机器学习能力的诉求，ADB 集成了 Spark 引擎，并在此技术上推出了 Servlersss Spark，其具备如下优势：
　　1.一份数据存储，在离线共享：无缝对接 ADB 已有元数据和数据；支持大吞吐读写 ADB 数据；Spark 批量写入的数据，在线分析查询可直接访问
　　2.数据库体系&体验：使用 ADB 统一的账号、权限和鉴权体系；支持通过 ADB Workflow、DMS 以及 DataWorks 调度编排 SparkSQL 作业
　　3.完全兼容 Spark 生态：基于最新的 Apache Spark 3.X 版本，充分享受开源社区红利；支持 SparkSQL、DataFrame API 主流编程接口以及 Thrift Server；支持 Spark UDF，支持 Hive UDF/UDTF/UDAF
　　4.按量计费，秒级弹性：开箱即用，按量计费无任何持有成本；基于神龙、ECS/ECI 的管控底座以及资源池化，缓存加速等技术，支持 Spark Pod 秒级拉起
　　对于实时性有诉求的场景，可以基于 ADB APS 服务可以非常方便的构建准实时 Lakehouse，白屏化操作快速配置入湖通道，多种数据源支持，满足不同数据源接入诉求，更多数据源也在持续集成中。
　　而对于实时性没有诉求的场景，可以基于 Spark + Hudi + ADB 工作编排构建离线 Lakehouse,如想对 RDS 数据构建离线Lakehouse进行分析，可使用ADB工作编排，利用 Spark 将 RDS 数据离线导入 Lakehouse，并做数据的清洗和加工，有需要最后可通过一条简单的 Spark SQL将数据从 Hudi 导入 ADB 做查询分析加速。
　　另外 ADB Spark 与 Hudi 和 ADB 表都做了深度集成，便于客户使用，如对于 Hudi 表的使用，免去了很多 Hudi 额外的配置，开箱即用；对于 ADB 表，可通过 Spark 创建、删除 ADB 表元数据，也支持读写 ADB 表数据。
　　另外最后介绍下 Spark 与 ADB 集成提供的 Zero-ETL 解决方案，这也与 2022 AWS reinvent 推出的数据集成服务 Zero-ETL 类似，我们通过一个场景了解 Zero-ETL 的应用及其优势。
　　客户如果对于 ADB 表有分析挖掘的需求，受限于 JDBC 方式吞吐的限制，需要先将 ADB内表数据以 parquet 格式导出到 OSS，利用 OSS 的带宽，再通过 Spark 进行分析挖掘，最后输出挖掘结果，可以看到这种方案中存在ETL操作，从而引入数据一致性差、时效性低的问题。
　　在 ADB 湖仓版中 Spark 与 ADB 做了深度集成，通过 Lakehouse API直接访问 ADB 内表数据，吞吐高，所以面对同样的场景，可以使用下面的链路，即直接通过 Spark 分析 ADB 数据，无需 ETL，数据一致性好，时效性也很高；另外对于 Lakehouse API 层的访问也支持列投影、谓词下推、分区裁剪等能力，更多下推能力也在持续建设中。
　　前面介绍了很多关于 ADB 湖仓版的功能以及优势，包括 Serverless Spark、APS 服务、融合引擎、工作流编排等等。而对于 ADB 湖仓版的定位总结成一句话就是从湖到仓，打造云原生一站式数据分析平台。
　　让客户通过 ADB 湖仓版平台就可以轻松玩转数据分析。
　　另外 AnalyticDB MySQL 湖仓版从 7 月份开始，历时 4 个月的邀测后，11 月 1 日正式开始公测，可以通过点击「AnalyticDB MySQL3.0婀栦粨鐗堝叕娴嬬敵璇 」进行公测申请。
　　今天的分享就到这里，谢谢大家。
　　分享嘉宾
　　李少锋
　　阿里云 数据库技术专家
　　Apache Hudi Committer & PMC成员，数据库OLAP ADB Hudi&Spark团队技术负责人。
　　DataFun新媒体矩阵
　　关于DataFun
　　专注于大数据、人工智能技术应用的分享与交流。发起于2017年，在北京、上海、深圳、杭州等城市举办超过100+线下和100+线上沙龙、论坛及峰会，已邀请超过2000位专家和学者参与分享。其公众号 DataFunTalk 累计生产原创文章800+，百万+阅读，15万+精准粉丝。