使用 LakeSoul 构建实时机器学习样本库

2022 年 6 月 02 日
本文字数：3815 字
阅读完需：约 13 分钟

首先，附上 Github 链接 LakeSoul：https://github.com/meta-soul/LakeSoul，可搜索公众号元灵数智，在底部菜单了解我们 - 用户交流获取官方技术交流群二维码，进群与业内大佬进行技术交流。

在之前的公众号文章《重磅！开源湖仓平台 LakeSoul 设计理念详解》中，我们介绍了 LakeSoul 开源流批一体表存储框架的设计理念和部分实现原理。LakeSoul 设计的初衷，是为了解决在流批一体的业务场景下，传统的 Hive 数仓难以解决的各类问题，包括 Upsert 更新、Merge on Read、并发写等。今天我们以一个典型的应用场景：构建实时机器学习样本库来展示 LakeSoul 的核心功能。

一、业务需求背景

1.1 在线推荐系统

在互联网、金融等行业，很多的业务场景都可以归纳为一个在线个性化推荐系统，包括搜索、广告、推荐、风控等。例如，在电商业务中，通过搭建个性化推荐系统，可以实现千人千面的猜你喜欢推荐，提升用户的点击率、购买率等；在广告业务中，个性化推荐是实现精准定向，提升 ROI 的核心系统；在金融风控领域，需要实现对用户偿还能力、逾期可能性的实时预测，为每个用户提供个性化的信贷额度、还贷周期等。

可以看到，推荐系统在各个行业领域都有着广泛应用。搭建一个工业级在线推荐系统，需要很多的环节和系统相互衔接，有比较大的开发工作量。元灵数智平台研发的 MetaSpore 框架提供了一站式的推荐系统开发解决方案，详细介绍可以看我们之前的公众号文章《重磅！基于新一代 MetaSpore 平台快速搭建工业级推荐系统》。

本文着重介绍如何构建一个实时的样本库，从而实现 “用户反馈 -- 模型迭代” 的完整闭环，让推荐系统能够自我学习迭代更新，快速捕捉用户的兴趣变化。

1.2 什么是推荐系统机器学习样本库

在推荐系统中，核心部分是一个个性化排序的算法模型。模型训练首先要构造样本，通过各类特征和用户行为反馈标签，去学习每一个用户的偏好。样本库往往包括几个部分：

用户特征（User Feature）：包括用户的基本属性，用户的历史行为和最近的实时行为等。其中用户基本属性特征可能是来自于在线的实时请求，也可能来自离线的 DMP 挖掘出的行为标签。用户的历史行为、实时行为一般包括用户历史上有过反馈行为的事件，以及相关的一些聚合统计指标；
物品特征（Item Feature）：物品是要给用户推荐的对象，可以是商品、新闻、广告等。特征一般包括物品的各类属性，包括离散的属性和统计值连续属性等；
用户反馈：即算法模型中的标签（Label）。标签是各类用户的反馈行为，例如展示、点击、转化等。算法模型需要通过特征和标签的关系，来学习建模用户的偏好。

1.3 机器学习样本库构建的挑战

在构建机器学习样本库时，往往会遇到这样几类问题和挑战：

实时性要求。业内主流推荐系统的模型学习已经往在线、实时化发展。模型更新越及时，对用户兴趣变化的捕捉就越快，从而能够给出更加精准的推荐结果，提升业务效果。这就需要样本库能够支撑很高的写入吞吐的能力。
多流更新。在线上通过模型进行排序计算后，会有大量的在线特征，需要实时回流供进一步模型训练使用。而用户反馈也同样需要回流到样本库，通常用户反馈会有多个实时流。在这种情况下，会有多个实时流，同时并发写入更新样本库的不同列。传统的 Hive 数仓一般是无法支持实时更新，需要通过全量 Join 的方式来实现，而在 Join 窗口较大的情况下，运行效率很低，还会带来大量的数据重复冗余。使用 Flink 的窗口 Join，也同样存在状态数据庞大，运行维护成本较高的问题。
并行实验。在实际业务开发中，算法工程师往往需要同时进行多组模型并行实验，以对比效果。不同的模型可能需要不同的特征以及标签列，这些列会通过不同的方式更新，例如一部分特征来自于离线的批量作业计算生成，这些批量数据也需要插入到特征库中。
特征回溯。在算法业务开发中，有的时候需要增加特征，而模型需要重新回溯。这就要求将新特征批量更新到历史的数据中。在 Hive 中也很难高效地实现。

可以看到，在推荐系统算法场景，构建一个实时的样本库，存在比较多的挑战。而这些挑战的主要问题是 Hive 数仓的功能、性能较弱，对流批一体、增量更新、并发写入等场景不能很好的支持。之前字节跳动等公司分享过基于 Hudi 的方案来进行流批一体构建推荐系统样本《字节跳动基于 Apache Hudi 构建实时数据湖平台实践》，然而 Hudi 在实际使用中仍然存在并发更新等问题。

由数元灵开发并开源的流批一体表存储框架，可以很好地解决以上这几类问题。下面我们详细介绍如何使用 LakeSoul 来构建一个工业级推荐系统的样本库。

二、构建实时机器学习样本库

LakeSoul 是一个为流批一体场景设计的表存储框架，具有如下几个关键特性：

行列级别更新（Upsert）
支持 Merge on Read，在读时进行数据合并，提高写的吞吐
支持对象存储，不需要文件语义
并发写入，可以支持多个流、批作业对同一分区进行更新
分布式元数据管理，提高元数据的可扩展性
Schema 演进，可以对表的列进行增删

通过 LakeSoul 来搭建机器学习样本库的总体设计是，使用 Upsert 代替 Join，将多组特征、标签分别通过流、批的方式写入同一个表，通过 LakeSoul 获得高并发的写入、高读写吞吐能力。我们来详细讲解具体的实现流程和关键原理。

2.1 主键设计

为了能够支持高效的 Merge，LakeSoul 提供了设置主键的功能。对于一个表中的主键列，会根据哈希分桶数，均匀的分片到指定个数的哈希桶中，而在每个桶内，对主键列进行排序后写入。在这种情况下，读时只需要对若干个增量文件按有序主键进行归并，就可以得到 Merge 结果。

在推荐系统样本库的场景，通常在线请求时会生成一个请求 ID，所有特征、标签的回流，都会带上这个 ID。实际在离线 Join 的场景，也是通过这个 ID 作为 Join Key。因此我们可以使用请求 ID 作为 LakeSoul 样本表的主键，并且以小时作为 Range 分区。我们可以通过下面的方式在 Spark 作业中创建 LakeSoul 表：

LakeSoulTable.createTable(data, path).shortTableName("sample").hashPartitions("request_id").hashBucketNum(100).rangePartitions("hour").create()

复制代码

这样就创建了一个以 request_id 作为主键，并且哈希分桶为 100，以小时作为 Range 分区的表。

2.2 数据写入和并发更新

由于特征、标签分别来自不同的流和批，我们需要多个流或批的作业并发对 sample 表进行更新。每一份数据都需要有 request_id 列和 hour 列，在执行 LakeSoulTable.upsert 时，LakeSoul Spark Writer 会自动根据 request_id 进行 repartition 分桶，并根据 hour 列写入对应分区的对应桶中，一批写入数据可以存在多个 Range 分区的值。

LakeSoul 支持多流并发的 Upsert，可以很好地满足样本库多流实时更新的需求。例如有两个流，分别是特征回流和标签回流数据，只需要执行 Upsert，就能够实时更新到样本库中：

// 读取特征回流，更新样本表val featureStreamDF = spark.readStream...val lakeSoulTable = LakeSoulTable.forPath(tablePath)lakeSoulTable.upsert(featureStreamDF)
// 读取标签回流，更新样本表val labelStreamDF = spark.readStream...val lakeSoulTable = LakeSoulTable.forPath(tablePath)lakeSoulTable.upsert(labelStreamDF)

复制代码

由于写入时并不需要进行 Merge 操作，只需要将当前的增量数据写入即可，因此写入可以有很高的吞吐，实际测试在云商对象存储上可以达到每个 core 写入速率 30MB/s 以上，即 30 个 Spark Executor，每个分配 1 个 CPU Core，就可以达到 1GB 的写入速度。

2.3 Merge On ReadLakeSoul

对于 Upsert 的数据，在读取时会自动进行 Merge。因此读取的接口和读一个表没有区别：

val lakeSoulTable = LakeSoulTable.forPath(path)lakeSoulTable.toDF.select("*").show()

复制代码

也可以通过 SQL Select 语句来查询。在底层实现中，对于每个哈希桶，由于主键已经有序，只需要进行多个有序表的外部归并，示意如下：

可以看到，样本流、标签流都分别执行了多次 Upsert，而某个时刻有一个读作业读取时，LakeSoul 会根据元数据服务中的更新记录，自动找到增量更新的文件，并执行有序外部归并。LakeSoul 实现了对 Parquet 文件的有序归并，并通过优化的小顶堆设计来提升多路有序归并的性能。

2.4 数据回溯（backfill）

由于 LakeSoul 支持任意 Range 分区数据的 Upsert，因此回溯和流式写入没有区别，将要插入的数据准备好之后，通过 Spark 执行 Upsert 就可以更新历史数据，LakeSoul 会自动识别 Schema 的变化，更新表的元信息，实现 Schema 演进。LakeSoul 提供了完整的数仓表的存储功能，每一个历史分区都是可以查询和更新的，相对于 Flink 进行窗口 Join 的方案，解决了中间状态不可见的问题，可以很方便的实现历史数据的大批量更新回溯。