业务背景

近年来，随着银行业务尤其是互联网金融业务的不断发展，金融业务数据量持续快速增长。同时，基于大数据、云计算、湖仓一体等技术体系的成熟，数据资产和价值挖掘得到越来越多的重视。G行自2019年开始便着手建设了以离线贴源数据为主的数据湖平台，实现了全行数据的统一存储、统一管理和统一服务。离线数据湖平台根据源系统数据表的业务属性区分处理后完成入湖入仓，并授权给下游租户进行批量运算或者数据查询，为业务营销、特征分析、个性化推荐、监管报送等提供了充分的数据基础。然而，随着银行业务场景和业务需求的不断发展，离线数据湖的不足也逐渐体现，主要包括：

1. 离线数据湖的数据时效性为T+1，业务应用基于T+1的数据进行批量计算，导致部分应用指标或业务数据相对滞后，而业务的发展要求实现数据价值的快速挖掘和分析，以便辅助业务需求快速迭代更新；
2. 离线数据湖平台以Hive作为数据存储和管理的主要组件，Hive数据处理性能较低，通常用于离线数据批量作业，无法满足具有较高时效要求的实时数据处理。同时，Hive不支持对数据的修改和删除，在需要对历史数据进行处理时，需要对全量数据进行重新覆写，极大的消耗队列资源和时间；
3. 目前行内已经基于kafka实现了对业务源系统增量更新数据的捕获和短时存储，然而不同业务系统对于该部分数据的处理和应用五花八门，对于数据的要求不一，部分数据重复消费和存储，浪费生产环境资源的同时也不便于精细化管理。

因此，为适应业务开展和数据分析的更高时效要求，保证数据资产的统一管理和应用，G行逐步建设了以准实时和实时数据为主的实时数据湖平台。

Hudi简介

Hudi（Hadoop Upserts and Incrementals）是目前主流实时数据湖技术之一，通过提供高效的增量更新、删除和查询能力，以及强大的数据索引和存储机制，解决了传统Hive场景下数据更新困难的问题，实现了对海量数据的快速插入、更新、管理和查询。Hudi具有如下特性：

1. 支持多种存储介质，如分布式文件系统HDFS、云存储S3、Azure等；
2. 提供COW（Copy on Write）和MOR（Merge on Read）两张表类型以及多种索引机制（如布隆过滤器等），可以实现T+0实时数据的快速读写；
3. 支持Spark、Flink、Hive等对数据进行增量更新、插入、删除等操作；
4. 自动管理小文件，通过合并小文件和删除无用文件减少文件系统的碎片化问题，保证查询性能。Hudi以其较高的数据摄取和处理效率使得大规模数据下的实时数据处理成为可能，数据更新插入和修改的效率大大提高，数据处理的时效性从T+1缩短至T+0。

图1. Hudi流式数据处理架构

Hudi中有两个重要的概念，时间线（Timeline）和文件布局（File Layout），如下图2所示。其中Timeline在Hudi中扮演着事务日志的角色，记录了表的所有操作（action）及其发生的时间点（instant time）和状态（status），提供了数据的即时视图和按到达顺序检索数据的能力。File Layout则描述了Hudi表在文件系统中的存储结构，Hudi中每个表都被划分为多个分区（partition），每个分区由多个文件组（File Group）组成，每个文件组由一个唯一的文件ID标识。每个文件组又包含多个文件片，每个文件片由一个基础文件（Base File）和一个或多个日志文件（Log File）组成。Base File是经过存储合并整理后的数据文件主体，而Log File记录了数据的增量变化，包括删除和更新操作等，读取时需要合并Base File和Log File中的数据获得最新数据。

图2. Hudi表Timeline和File Layout示意图

实时数据湖架构实现

实时数据湖平台基于Hadoop大数据平台进行建设。湖内数据主要来源于支撑银行各项金融业务开展的联机业务系统及业务支撑类系统。源系统通过批量调度和CDC（变更数据抓取）方式分别实现批量数据和流式数据的采集和推送。如前所述，Hudi与Hadoop生态体系中众多组件是相互兼容的。Kafka中的流式数据在通过flink进行消费后可以快速地以hudi格式进行写入，底层仍基于HDFS实现数据存储和备份管理。整体架构设计如下图3所示。

图3. G行实时数据湖架构

源系统数据通过CDC工具实现增量数据更新获取，并写入Kafka集群。数据在写入Kafka前，首先会与Schema Registry服务进行交互。Schema Registry主要用于存储和检索Avro、JSON和Protobuf等序列化格式的模式（schema），通过RESTFul接口提供服务，使得kafka的生产者和消费者能够方便地注册、检索和使用schema。Flink在消费kafka数据前，与Schema Registry通信获取对应的schema来反序列化消息内容，还原为原始的数据格式。

3.1 元数据管理服务

目前Hudi社区的实现是使用分布式存储（HDFS）来管理Timeline和表的基础元数据信息（如schema、分区信息、数据文件路径等）。在分区数量比较大时，元数据的获取需要从 HDFS 拉取多个文件，在读表的时候需要拉取大量目录和 Timeline上记录的表对应的操作进行比对，找出最新的版本包含的文件，这样导致在近实时的场景下带来了比较大的时延。通过集成Hive Metastore，可以将Hudi表的schema、分区等信息同步到Hive，使得其他支持Hive的系统能够更方便地查询Hudi表中的数据，提高了查询性能，但同时也可能存在延迟，影响数据的实时性和一致性。

为了实现元数据的统一和更加高效地元数据管理，我们引入了Hudi MetaServer作为统一的元数据管理服务。如下图所示，Calalog Service提供了统一的元数据访问入口，当Hudi Client提交数据更新或者表结构变化请求（Commit）的时候，Request Handler会将该请求路由到Service层进行处理，Timeline Service修改提交状态并更新Timeline的action和status信息，Partition Service根据数据分区变化更新元数据里的分区信息，Snapshot Service更新元数据中snapshot的增删改内容，元数据信息均写入到可插拔的底层存储如MySQL、KV数据库等，完成元数据的更新。通过EventBus和Listeners可以实现表结构等元数据更新变动的监听和订阅，并且同步更新到Hive Metastore，实现不同元数据管理组件之间的数据同步。

图4. Hudi元数据管理流程

在处理元数据访问或查询请求时，Request Handler会将请求路由到Table Service拿到表的信息，在对表进行SQL计算执行优化的时候，去请求Partition Service根据分区信息扫描文件，同时基于Timeline Service返回的Timeline信息调用Snapshot Service拿到对应分区文件，实现文件的快速读取。

通过Hudi MetaServer，实现了对元数据的高效管理，支持HDFS、KV、MySQL等存储介质，解耦了元数据存储，提高了元数据的写入和查询效率，而且扩展了文件级的元数据管理和并发更新，更加适配大量快速实时数据的读写要求，同时，可以兼容Hive Metastore，对外可以提供统一的元数据视图和数据一致性保障。

3.2 表管理服务

如前所述，Flink在消费Kafka数据写入Hudi表时，会产生很多的增量文件（Log File）。在流式数据处理中，数据并行写入多个小文件可以提高写入吞吐量，但读取时更希望访问少量的大文件来提高检索效率。Hudi提供了Compaction和Clustering机制来实现减少文件碎片，优化存储结构，加快数据读取速率的目的。其中，Compaction主要用于在MOR类型表中周期性地将基于行的Log File合并生成新的基于列的Base File，Clustering则是通过周期性的将小文件合并成大文件，从而减少存储碎片和元数据管理开销，提高查询性能。

针对这些内置到引擎内部的任务，管理不当会对流式任务的稳定性有很大影响，甚至可能造成数据写入任务的延迟或冲突等。因此，我们将表管理服务（Table Management Service，TMS）独立出来，通过异步方式实现对这些内置任务的集中管理，降低文件合并等任务对流式数据写入的影响，提高Hudi写入稳定性和查询性能。

表管理服务实时同步Hudi MetaServer的表事件（Event）信息，并且会根据Event信息和既定策略生成任务排期计划，交由调度服务（scheduler）和任务管理器（Job Manager）去执行定时任务动作，一般由Job Manager启动一个Spark或者Flink任务执行，执行过程中会监控和记录作业的状态和相关信息，并且支持任务失败自动重调。同时，调度服务会根据资源队列情况进行动态计算和自适应调度，同时控制并发度以避免任务对资源队列的过度侵占。

图5. Hudi表管理服务流程

表管理服务通过异步方式实现了对Hudi表的优化存储和HDFS小文件的治理，保证了实时数据入湖主任务的性能。同时，对用户屏蔽了数据湖底层技术细节，让用户可以更加关注Hudi功能应用，而非底层技术优化，以一种更友好的方式保障了Hudi功能的可靠性和稳定性。

3.3 任务管理服务

任务管理服务主要用于管理实时数据消费任务的启停和状态监控。在创建Flink消费任务时，基于Hudi表名或者Kafka topic名称对Flink任务进行命名，如此，可通过任务管理台实现对具体某一张Hudi表或者某一Kafka topic消费任务的状态查询、任务阶段查询、日志详情查询，更方便对实时数据消费任务的管理和运维。同时，我们对Hudi相关功能接口进行了监控，以便于及时发现Hudi的异常，保障实时数据流任务的正常运行。

图6. Hudi任务管理

图7. Hudi接口QPS统计

图8. Hudi接口异常监控

任务管理页面方便了实时数据湖平台的运维和管理，通过持续完善和优化页面功能，可以辅助运维或开发人员进行更好的运营分析和日志排查，进一步加强对平台的管理。

总结与展望

G行基于Hudi构建了实时数据湖系统，并且实现了对元数据管理服务、表管理服务等的管理优化和提升，保障了实时数据湖的可用性和稳定性。在此基础上，G行也将持续完善实时数据湖架构及功能，实现流批一体的数据湖体系建设。同时，基于业务需求实现更为精细化管理，保障银行业务的开展和数据资产归整。

作者：郭涛

从事大数据运维工作多年，目前主要负责G行数据湖平台的运维和应用管理。

编辑：邸顺帆

正在成长中的应用管理员，爱好养多肉、游泳，积极工作，乐观生活。

G行基于 Apache Hudi 的实时数据湖架构与实践