字节跳动基于 Hudi 的机器学习应用场景
动手点关注
干货不迷路
本文为ApacheHudi技术社区分享会第十期嘉宾分享文章，主要介绍火山引擎LAS团队自研的多场景样本离线存储技术，用于处理机器学习系统的离线数据流。同时，还会为大家揭秘流批一体样本生成的过程，分享对Hudi内核所做出的优化和改造，探索其在数据处理领域的实际应用和效果。文末更有专属彩蛋，新人优惠购福利，等着你来解锁！
本篇文章提纲如下：
业务场景
离线样本存储与迭代
流批一体的样本生成
功能与优化
1.业务场景为了让大家更容易理解接下来要讲的基于数据湖的样本存储和样本生成问题，文章先给大家简单介绍一些相关的基础概念。首先是机器学习系统的离线数据流架构，机器学习系统和其他线上服务系统类似，其中和样本有关的角色也比较集中。如下图所示，整个离线数据流架构分为流式和批式两种类型，其中的样本数据由两部分构成，分别是特征和标签。
在流式架构中，特征由在线预估服务在serving时dump对应的快照并发送到消息队列中。标签则来自实时行为采集服务，通过日志上报等方法采集得到。在线样本生成服务消费两个数据流，通过关联得到完整的样本，并发送到下游的流式训练服务中进行模型训练，完成样本数据的消费。
批式架构是流式架构的补充，批式架构在订阅流式数据的同时，还会加入批式的特征或者批式生成的标签。比如风控反作弊或者广告类的业务，会有批式生产的数据，并使用批式的样本生成模块生成样本，进而被模型训练组件消费。
流式和批式数据流架构中，还有元数据服务，元数据服务记录了特征的相关元数据，流式批式数据流都会访问元数据服务获取meta信息。因此，我们对于批式的特征存储有若干种特定的访问pattern。
读方面有以下读数据pattern：大范围的按天批式读取，关注吞吐指标；秒级的点查；高效的谓词下推查询能力；存在基于主键/外建的join。
在写方面需支持以下能力：基于主键的upsert；针对部分cell的插入与更新；针对行/列/cell的删除；基于外键的upsert。
在这样的背景下，我们了解Hudi在机器学习离线数据流中的若干应用场景。
2.离线样本存储与迭代我们希望设计的样本离线存储方案能够适用于多种场景，主要包含以下三类情况。
第一，模型的重新训练，回放流式训练的过程，迭代/纠偏模型等等。
第二，样本的数据迭代，增加修改或者删除对应的特征/标签，并重新训练模型。
第三，样本的OLAP查询，用于日常debug等。
为了能够支持以上的场景的样本存储与迭代，我们提出的存储方案整体架构设计如下。在逻辑建模上，构建样本存储和构建特定pattern的Hive表非常类似，样本包含主键、分区键、内部元数据列等功能性column，然后包含若干特征列和若干标签列。在物理架构上，通过流式和批式生产/采集的特征数据和标签数据通过多个作业混合upsert的方式写入Hudi，更新位于KV存储的索引信息，并将实际的数据写入HDFS中。由于Hudi基于主键/外键upsert的特性，数据会被自然地拼接在一起，形成完整的包含特征和标签的样本数据，供消费使用。
在对离线特征进行调研时，我们需要面临以下挑战：基于HDFS这种不可变的文件存储，如何实现低成本低读写放大的数据修改。在没有使用数据湖之前，用户做离线特征调研之前需要复制样本，修改并另存一份。其中消耗了巨大的计算和存储资源，伴随样本量的增大，这样的方案将消耗数个EB的存储，使得迭代变得不可能。
我们基于Hudi实现了ColumnFamily的能力。这个方案受到了经典BigTable存储ApacheHBase的启发，将IOpattern不同的数据使用不同的文件进行存储，以减少不必要的读写放大。原理是将同一个FileGroup的不同列数据存储在不同的文件中，在读时进行合并。这种方法会将新增列的数据单独进行文件存储，发生修改或者新增成本很低。
我们通过为调研特征列赋予单独的CF的方式来减少读写放大，其他列复用线上的特征所在的CF。这样资源的使用量只会和新增特征相关。这种方式极大得减少了迭代所需的存储使用，并且不会引入任何shuffle操作。
上文介绍了离线样本的存储与迭代方案，接下来我们进一步为大家介绍在线样本生成时的流批一体生成方案，讨论其如何降低在线存储的使用成本。
3.流批一体的样本生成在线样本生成服务中，我们使用KV或者BigTable类存储来满足样本拼接的需求，比如RocksDB等。这类存储点查性能好，延迟低，但是存储成本也较高。如果在数据有明显的冷热分层的情况下，这类存储本身并不能很好的满足这样的存储需求。Hudi是一个具有KV语义的离线存储，存储成本较低，我们将冷数据存在Hudi上的方式来降低在线存储的使用成本，并通过统一的读写接口来屏蔽差异。这一架构也受到了目前市面的多种HSAP系统的启发。
为了能够让Hudi支持更好的点查，我们复用了写时的HBase索引。点查请求会先访问HBase索引找到数据所在文件，然后根据文件进行点查。整体端到端的延迟可以做到秒级。适合存储数据量大，qps较低的场景。
4.功能与优化在使用Hudi满足诸多业务需求的过程中，我们也对其内核做了一些改造，以更好得服务我们的业务场景。
4.1LocalSort我们支持了单文件内的主键排序。排序是较为常见的查询性能优化手段。通过对主键的排序，享受以下收益
CF在读时，多CF合并使用SortMerge的方式，内存使用更低。
Compaction时支持SortMerge。不会触发spill，内存使用低。我们之前使用SSD队列来做Compaction以保证性能，现在可以使用一些廉价的资源（比如无盘的潮汐资源）来进行Compaction。
在流批一体的样本生成中，由于主键是排好序的，我们点查时基于主键的谓词下推效果非常好。提升了点查性能。
4.2Bulkload并发写并发写一直是Hudi的比较大的挑战。我们的业务场景中会发生行级别/列级别的写冲突，这种冲突无法通过乐观锁来避免。基于机器学习对于数据冲突的解决需求，我们之前就支持了MVCC的冲突解决方式。更进一步得，为了能够让Hudi支持并发读写，我们参考HBase支持了Bulkload的功能来解决并发写需求。所有写数据都会写成功，并由数据内部的mvcc来决定数据冲突。
我们首先将数据文件生成到一个临时缓冲区，每个缓冲区对应一个commit请求，多个写临时缓冲区的请求可以并发进行。当数据完整写入临时缓冲区之后，我们有一个常驻的任务会接受数据load的请求，将数据从缓冲区中通过文件移动的方式load进Hudi，并生成对应的commit信息。多个load请求是线性进行的，由HudiTimeline的表锁保证，但是每个load请求中只涉及文件的移动，所以load请求执行时间是秒级，这样就实现了大吞吐的数据多并发写和最终一致性。
4.3CompactionService关于Compaction，Hudi社区提供了若干Compaction的开箱即用的策略。但是业务侧的需求非常灵活多变，无法归类到一种开箱即用的策略上。因此我们提供了CompactionService这样的组件用来处理用户的Compaction请求，允许用户主动触发一次Compaction，并可指定Compaction的数据范围，资源使用等等。用户也可以选择按照时间周期性触发Compaction，以达到自动化数据生效的效果。
在底层我们针对Compaction的业务场景做了冷热队列分层，根据不同的SLA的Compaction任务，会选择对应的队列资源来执行。用来降低Compaction的整体成本。比如每天天级别的数据生效是一个高保障的Compaction任务，会有独占队列来执行。但是进行历史数据的单次修复触发的Compaction，对执行时间不敏感，会被调度到低优先级队列以较低成本完成。
针对数据湖的样本存储与生成问题，我们搭建了适用于多种场景的存储方案架构，实现了批流一体的样本生成，并且通过对Hudi内核进行一定的改造，实现更加满足实际业务需求的功能设计。
以上就是字节跳动在Hudi的实践，目前均已通过火山引擎湖仓一体分析服务LAS产品对外服务，欢迎对这方面有需求、感兴趣的用户都可以积极地来体验一下我们的LAS湖仓一体分析服务。
湖仓一体分析服务LAS（LakehouseAnalyticsService）是面向湖仓一体架构的Serverless数据处理分析服务，提供字节跳动最佳实践的一站式EB级海量数据存储计算和交互分析能力，兼容Spark、Presto生态，帮助企业轻松构建智能实时湖仓。新人优惠来袭！赠送给所有新人用户的专属福利来啦，LAS数据中台新人特惠1元秒杀活动最新上线！更有超多叠加优惠等你来抢！感谢大家一直以来对我们的支持与厚爱，我们会一如既往地为您带来更好的内容。
（点击文末“阅读原文”，可顺滑体验）