靳国卫：快手如何使用Hudi解决数据内容建设的效率问题？

       导读：从领域模型建设的角度，我们知道需要对业务场景进行合理的拆解、规划、设计，使得我们在进行数据内容的建设时思路清晰、模型复用度高。然而在实际的实施过程会遇到一些复杂场景，在我们极尽优化之后仍然出现就绪晚、执行效率低的情况。所以，如果当前的技术手段不再能解决这些问题，我们会分出一些精力去探索新的技术方案来解决这些效率问题。

       01痛点业务场景

       1. 痛点场景

       (1) 数据调度

       日常工作中有些业务周期不是自然周期，比如我们的花费计算场景是14点启调，调度启动晚。业务上可能更新近3个月的局部数据，需要动态回刷3个月的数据，计算周期长。调度启动晚和计算周期长，造成就绪比较晚，影响业务对数据的使用诉求。

       (2) 数据同步

       每天会同步大量的过程明细数据到数仓进行计算，用于支持业务决策。有些场景需最新状态，比如我们有一份用户关系数据，只需要最新的关系对。还需要将关注过程做合并计算。过程数据量大和合并计算耗时长，造成完成晚，并且同步任务是链路根节点，导致整条链路的SLA保障压力大。

       (3) 修复回刷

       进行历史数据回刷时，其实仅需要更新小部分数据即可，然而离线数仓是全量批更新模式，需要耗费大量的资源做全量回刷。因此回刷修复会耗费大量资源、数据修复周期长。

       2. 基于痛点场景的思考

       从业务痛点出发，不论从研发的角度还是业务的角度其诉求一致：数据能够快速就绪、产出最新的数据。

       为什么当前的离线数据建设技术方案不能满足需求？从本质来看，仍是采用传统的调度+离线计算的方式，只不过采用了资源换效率的方式，改变的仅仅是调度的频次。

       根据数据特点 + 【产出快】【状态新】的诉求，也做过一些尝试。

       调度场景，因为只有千万级的3个月数据，可以全部保存在Flink的内存中，以天为窗口Sink一次，这样可以实时获取最新数据。但随着业务增长，资源占用、稳定性保障都面临较大挑战。

       同步场景，将每小时数据进行同步，同时完成小时内过程数据的合并，再同前一个小时的全量历史数据合并生成小时级别的全量数据。这样时效性有所提升，但存储和计算的成本是天合并的数倍。

       我们希望能在生产时像业务系统那样，实时完成业务数据处理，分析拥有离线大数据的强大计算能力，希望可以对离线数据模型进行增删改查，拆解后就是【实时化计算】+【离线数据模型】的CRUD。

       02为什么选择Hudi来解决问题

       为了实现【实时化计算】+【离线数据模型】的CRUD的诉求，业界的解决方案不止一种，我们选择Hudi基于以下原因：

       功能丰富度：更多的功能才能支持更多的场景，从而解决更多的业务痛点问题；

       公司融合度：数据内容建设依托公司的体系架构，与公司的体系架构匹配，才能随着公司体系架构建设，让数据内容建设的效率更高、保障更好；

       自动化程度：数据内容建设的核心目标是高效的解决业务问题，因此方案的自动化程度越高，我们就可以抽更多的时间聚焦于业务问题的解决，而不是技术细节本身；

       Flink集成度：快手的实时化计算的解决方案以Flink为主；

       社区活跃度：整体来说数据湖的应用还处于比较初期，因此遇到问题可以在社区讨论，集合社区同学智慧，获取建议、经验，从而高效的有效的解决问题。

       Hudi可以从以下方面解决痛点问题：

       从实时数据源Binlog & Kafka采集数据，通过Flink & Spark Streaming进行数据处理，满足实时化计算的诉求；

       计算过程中可以实现对离线表进行增删改生成Row Tables，同时可以被离线数仓识别和应用，满足离线数据模型CRUD的诉求。

       从上图中一条数据的Hudi之旅我们看下数据的计算过程。

       从实时源采集数据到系统后，首先做数据的打散平均重分布，从而避免因为业务热点或者分区数据不均匀造成的数据倾斜和长尾问题；

       经过重分布之后在每个节点上就可以拿到均匀的分区对应的数据；

       经过Shuffle之后可以将相同分区的数据分发到相同的节点，此过程和hudi的设计有关，hudi是一个分布式集群计算，它最终在数据刷盘的时候是按照分区进行pipeline计算的，这时候如果多个物理节点存在相同分区就会造成写锁的抢占影响效率；

       在此过程中我们可以对数据做合并策略优化，相同key的数据进行合并，只保留策略诉求的数，从而减少hudi的计算量；

       拿到了需要更新的数据，我们要知道哪条记录更新到哪个文件中就需要用到索引检索服务，合理的选择索引可以提升数据与文件的匹配速度；

       最终将记录写入文件的时候还可以添加丢弃策略，比如文件中已经有更新版本的数据了，就可以丢弃本次记录，从而进一步提升写效率。

       经过一系列的操作和优化之后，我们就可以将数据写入到集群，之后可以通过Hive的元数据挂载成普通表进行使用。

       对于存储结构，Partition的目录结构就是我们的hive分区结构，Base是我们的Hdfs存储路径。

       在了解Hudi的架构(骨架)、计算过程(大动脉)、存储结构(细胞)之后，我们会思考Hudi究竟是什么呢?

       初衷是为了解决离线模型更新的问题，以存储的方式来实现，因此是一个数据存储解决方案；

       为了实现数据更新需要用到计算，Hudi借用了当前成熟的计算方案来实现自己的架构，所以算是集成了计算引擎。

       03如何使用Hudi解决业务问题

       简单回顾一下前面提到的痛点：

       计算数据量大 + 计算时间长 + 同步任务是链路根节点，造成SLA压力大

       回刷周期长 + 技术方案单一，导致资源浪费严重、修复效率低

       1. 挑战一：基于Hudi的模型设计

       我们的诉求是使用Hudi来解决时效性和效率问题，然而Hudi和离线的模型设计是不一样的，要避免思维的惯性和认知的偏差。

       比如，数仓分区设计：

       让数据的组织形式更加匹配业务形态，在使用的时候更加易于理解，易于使用

       在技术层面，分区用来做数据裁剪，进而提升查询效率

       小文件合并优化：

       当前hdfs的架构设计下，系统过多的小文件会影响集群的稳定性

       小文件太多会启动过多的任务，从而降低集群的任务并行度

       然而，在Hudi场景下，除了数仓的模型设计规范，还需要考虑以下几点。

       数仓分区设计：

       分区与Hudi写数据的并发有关

       分区是解决写倾斜的处理手段之一

       小文件合并优化：

       文件越小更新效率越高

       早期Hudi更新单个数据文件的时候是顺序的，如果文件过多等待的排队时间会变长，要取舍排队和更新时间的平衡

       从我们的痛点出发，设计的时候会更多的考虑写的效率，因为写数据要与磁盘交互，效率比较低，所以设计思路有两点：减少数据量和提升写效率。

       基于减流量、写效率的指导思想，我们基于Hudi的数据模型设计会考虑以下几方面。

       主键&分区设计

       - 主键是确定数据记录和对其进行增删改的根本依据，因此需要深刻的理解业务并以此为依据设计合理的稳定的主键，否则当主键改变时，需要重新构建数据，成本较高；

       - 主键是索引的生成依据，当主键改变时，需要重新构建数据；

       - 分区决定了数据写到哪个目录，在哪个目录内更新，如果分区的设计发生改变，相同主键的数据会被写入到不同的目录中，造成数据查询时数据重复，当分区改变时，需要重新构建数据；

       - Hudi的分区数量可以用来控制pipeline的并发量。

       合并&索引策略

       - 通过数据的合并策略可以减少Hudi处理的量，从而提升处理效率；

       - 索引策略是定位key与文件关系的依据，快速的定位策略可以更快的完成文件查找，提升Hudi效率。

       并发&丢弃策略

       - 分区是提升并发的手段之一；

       - 在真正写文件时，Hudi会对比待处理数据与已存在数据的的状态，这样乱序、重复数据直接被丢弃掉。

       基于写模型设计，在数据同步的场景，可将数据模型设计如上图。

       2. 挑战二：集中毛刺更新

       快手会在节假日做活动，这时候关系数据就会成倍于日常的增长，预热期间数据开始爬坡式增长，活动期数据增长速度达到顶峰，活动结尾数据增长变缓。

       基于三段式的增长趋势，在实现上我们也面临了三个阶段的挑战：数据开始增长，程序处理压力变大，收到预警；随着预热进行数据持续变大，程序的处理能力出现瓶颈，任务开始失败；数据持续变大积压，同时失败原因是处理失败，所以即使重启也不能完成数据处理，造成需求支持失败。

       针对以上三个阶段的问题做了以下应对：

       减少Hudi数据处理量

       - 通过业务数据分析发现，在活动期间有用户参加活动后取消关系的操作，设置合并策略可以减少Hudi的数据处理量；

       - 通过设置丢弃策略可以减少乱序、重复数据的处理，提高Hudi的写文件效率。

       提升索引的效率&稳定性

       - 选用分区布隆索引，key与文件的映射关系存储在文件中，系统重启、任务发布时可保证数据正确性；

       - 通过分区个数控制单分区的数据量，因为单分区的索引大小可控，索引加载效率可控；

       - 索引大小可控，索引在内存中占用的比例可控，可提升执行程序的稳定性。

       提升写效率

       - 第一阶段为解决并发问题， 采用user_id做mod来控制并发度，让程序在平峰时可以稳定运行；

       - 第二阶段发现个别分区有明显长尾效应，发现是大V分区数据倾斜，所以采用user_id + follow_id做mod来控制并发，解决倾斜长尾问题；

       - 早期为提升写文件效率设置单个小文件大小为128M，后来出现长尾效应后发现Hudi在执行数据更新时在一个分区内是一个个文件排队更新的，这时候如果小文件过多会出现排队时间较长的情况（新版本已经可以设置单分区的写文件并发量）；

       - 需要平衡设计文件更新时间和文件执行排队时间的最优解。

       分区的改变需要重新构建数据，所以一定要在模型设计时深刻的理解业务做好规划，尽量不要出现边开飞机边换发动机的情况。

       3. 挑战三：数据回刷局部更新

       已经解决了基础数据大数据量更新的复杂场景，还希望在处理数据回刷局部更新问题时，能解决研发效率问题，形成一套标准化的可以复用的解决方案。

       诉求是有一份业务数据，包含日期分区、主键设备ID、渠道属性，在01~03号的时候渠道为SEM，在04号的时候希望将数据更新为PinPai。

       经过探索后，我们最终形成了离线 + 实时 + Hudi的通用解决方案。通过回刷补数的案例来说明。

       因为是补数场景所以暂时不需要所有的例行都是用实时的方式生成Hudi可识别的数据结构，因此日常生产可以通过离线批处理的方式生成Hudi可识别的存储结构就可以。当需要进行数据修复的时候生成需要更新的记录即可(Hudi目前是按行整体更新的)，比如右上图的修复记录。将修复数据传递给实时的Hudi处理程序，对结果数据直接更新即可。修复数据是通过更高的版本来覆盖例行版本，比如例行版本号为0，修复版本号为20210503，其原理是合并&丢弃策略。对于对时效要求高的需求，可以直接从实时数据源采集数据，通过Hudi实时处理，生成下游可用的结果表。

       通过以上的架构我们可以实现对高时效、补数回刷的场景的覆盖，同时也沉淀了一套通用的模板，进而提升了整体的研发效率。

       4. 挑战四：Hudi保障

       通过数据同步、补数回刷场景的实践，Hudi已经解决了我们的业务痛点，也可以提升我们的研发效率。然而初期业务上线时仍有一些担心，比如如何及时发现问题、定位问题、确保数据准确性？我们通过如下方法来提供保障。

       时效性：将整个数据流程分成6个阶段5个流转过程，当收到监控预警时可以根据预警的流转阶段第一时间识别哪个阶段哪个流转出了问题，从而能第一时间定位问题发生在哪里，进行问题排查；

       准确性：通过探针每隔10分钟从业务数据抽取10条数据与Row Tables做对比，如果95%一致则通过、90%以上介入验证、90%以下则进行业务阻断，人工确认问题或修复问题后再恢复任务；

       流程&设计：Hudi的模型设计和数仓的模型设计有一些差异，会对模型设计进行严格评审，对开发、测试、上线、监控配置流程做严格的审查。