响应式网站空间服务器要求,怎么注册地理位置,山西省消防总队建设工程备案网站,花生棒做网站01 背景 推荐系统的模型#xff0c;通过学习用户历史行为来达到个性化精准推荐的目的#xff0c;因此模型训练依赖的样本数据#xff0c;需要包括用户特征、服务端推荐的视频特征#xff0c;以及用户在推荐视频上是否有一系列的消费行为。
推荐模型数据流#xff0c;即为…01 背景 推荐系统的模型通过学习用户历史行为来达到个性化精准推荐的目的因此模型训练依赖的样本数据需要包括用户特征、服务端推荐的视频特征以及用户在推荐视频上是否有一系列的消费行为。
推荐模型数据流即为推荐模型提供带特征和优化目标的训练样本包括两个模块一是Label Join模块负责用户行为的采集。二是feature extract模块从原始日志中抽取特征并基于用户行为计算模型优化的目标label。
在B站早期的推荐模型数据流架构中如下图所示采样两阶段特征补齐设计。Label Join模块除了完成用户行为的采集还需要查询实时特征补齐训练样本依赖的部分原始特征数据一般是秒级更新的实时特征存储在Redis中。而Feature Extract模块在计算样本之前补齐另外一部分原始特征数据一般是批量更新的特征数据存储在KFC中B站自研的KV系统 02 问题分析 上述的推荐模型数据流架构存在“不一致”问题包括“数据不一致”和“计算不一致” 2.1 数据不一致 在这种自反馈系统中推理输入的特征版本和训练输入的特征版本如果有差异的话会影响模型的准确性。数据不一致有3个原因 访问时间差异。推理服务和Label Join/Feature Extract模块访问同一个特征的时间是不一样的。特别是秒级更新的实时特征这种数据差异会被放的更大 新稿件问题。在线推理服务有稿件服务可获取新稿件的特征。而离线没有这个服务所以离线缺少新稿件的特征导致新稿件的推荐不准确 特征穿越问题。LabelJoin有N分钟的固定时间窗所以Feature Extract在查询批量特征的时候可能批量特征版本已经更新查到的是最新版本特征从而导致特征穿越。一般需要有经验的算法工程师在离线批量特征更新上加上时间延迟规避穿越问题。但这种规避可能会进一步加剧数据不一致 2.2 计算不一致 这里“计算”指的是从原始数据生成特征的过程。特征可以用于在线推理和离线训练。特征计算有3个地方 在线推理一个c的服务。特征计算使用c实现 Online Feature Extract实时的样本计算基于FLINK特征计算使用java实现 Offline Feature Extract离线样本回溯基于Spark特征计算使用python实现
此外这3个地方的数据格式也是不一样的不是简单的语言之间的转换。需要用户开发3种特征计算逻辑并且要很小心的对齐 03. 一致性架构 为了解决上述的“数据不一致”和“计算不一致”我们将B站推荐模型数据流升级成一致性架构: 数据一致性将在线推理的原始特征现场snapshotdump到近线。近线基于snapshot做Label Join和Feature Extract。因为在离线用的同一份数据可保证数据完全一致 计算一致性特征计算逻辑算子化基于tenforflow实现一个c lib。推理服务直接调用特征抽取lib离线Feature Extract通过java JNI调用c lib。特征计算都基于同一个c lib用户只需要定义一次可保证计算完全一致 3.1 整体架构 -推荐服务将推理的原始特征现场打包成snapshot通过数据集成工具从在线服务同步到近线Kafka
-在近线基于flink latency join功能实现Label Join为每条请求join用户在稿件上的各种行为labele.g. 点击、播放、后验等
-Label Join任务输出Shitu同时写到kafka和hive
-在近线基于flink实现实时样本计算消费kafka Shitu产出训练样本写到kafka再接入实时训练。产出模型用于线上推理
-在离线基于flink/spark实现的批量样本计算从hive表load Shitu产出样本写到hive。训练任务依赖hive表作批量的模型训练 3.2 Label Join Label Join使用flink letency join延时拼接功能完成每个请求稿件上的用户行为采集。用户行为作为训练样本的label即模型的优化目标。
flink letency join基于时间驱动以一个固定时间窗做数据下发。e.g.固定时间窗是N分钟即每条请求数据到flink任务后等待N分钟后输出数据
在最新的一致性数据流上我们在Label Join上增加了事件驱动增加数据下发的时效性将数据流的时效性提升60% 定义下发事件原则上按用户不会再看到视频作下发规则。对用户连续请求按时间排序。如果最新的请求是清屏请求那这次请求前的所有请求都可以下发。如果是普通请求那这次请求往前第5个请求可以下发 采用事件驱动时间驱动的数据下发方式优先事件驱动下发没有被事件触发的请求走时间驱动超时下发 3.3 样本计算 样本计算是基于Label Join产出的Shitu数据计算训练样本分两种模式
-online extract实时的样本计算flink streaming计算引擎读kafka写kafka
-offline extract离线的批量样本计算支持flink batch/spark batch两种计算引擎。offline extract支持两种样本计算模式1) 无新增特征的样本计算直接读Shitu hive table产也训练样本。2) 有新增特征的样本计算用户挖掘的新特征不在Shitu里。训练样本依赖Shitu和新挖掘特征
目前一致性的样本计算框架支持两种模型
1直接计算一般用于精排模型。整个样本计算过程抽象成几个算子
-selector数据筛选。过滤请求或者稿件
-calculate label : 通过用户行为label计算每个视频的train label
-刷内item采样在一刷请求内对稿件进行采样e.g. 按正负例
-pyfe调用fealib生成模型特征
每个算子都可以支持算法同学根据业务需求自定义 2有外部采样的样本计算一般在召回模型上使用
-calculate label : 通过用户行为label计算每个视频的train label
-外接一个采样稿件候选池根据稿件的train label进行采样。采样逻辑按算法需求可定制
-从KFC查询采样稿件的特征并组装一条完整的snapshot
-pyfe调用fealib生成模型特征 3.4 BackFill BackFill特征回填指的是算法同学调研新特征在模型上的收益流程如下 对于NoDelta模式直接读Base Shitu生成全量的训练样本 对HasDelta模式用户挖掘一批新增的特征delta snapshot) 基线Shitu join delta snapshot生成一份新Shitu 基于新Shitu作全量的特征计算生成全量训练样本 模型训练样本并评估auc效果不符合预期重新设计数据和特征
同时我们提供了一套python sdk支持用户在镜像或者jupyter上自己订制特征回填特征的逻辑和流程 3.5 基于protobuf wireformat的partial decode优化 对于在线推理现场snapshot采用了protobuf组织数据包含了模型特征需要所有原始数据单条数据超过250KB有上千个字段。在样本计算阶段对snapshot有两个处理逻辑
调用protobuf ParseFrom接口将snapshot bytes 反序列化成Message平均耗时7~8ms
将snapshot所有稿件类的特征做裁剪一刷请求n个稿件其中m个稿件参与训练平均耗时5~6ms
通过性能分析样本计算中有50%的时间消耗在上述snapshot protobuf解析和处理上。但实际样本计算相关逻辑上并不需要所有snapshot字段所以我们使用protobuf wireformat对snapshot做partial decode只解析需要的field。最终将snapshot处理的性能从14ms优化到1.5ms样本计算的cpu资源降低了30% 04 未来工作 4.1 基于Iceberg批流一体的训练样本计算框架 如3.1章节的数据流架构中通过FLINK实时计算产出的训练样本会同时写到Kafka和Hive表分别用于实时训练和批量训练。同时离线回溯也可以产出训练样本写到Hive表。这种架构存在两个问题
(1) 需要额外的FLINK资源把Kafka中的样本备份到Hive表中即一个实验样本流需要搭建两个FLINK任务
(2) 实时样本和离线样本输入输出的介质不同框架层面需要适配。下游训练模块也需要适配不同的样本源无法做到批流一体
未来我们计划引入iceberg实现样本计算框架的批流一体解决上述问题。Apache Iceberg 是一种用于大型分析表的高性能格式旨在解决数据存储和计算引擎之间的适配问题其核心特性之一是支持同时处理流数据和批数据提供统一的读写接口
框架如下图所示 Label Join产出数据实时写到iceberg Shitu表 样本计算框架从iceberg Shitu读数据可以实时计算或批量计算产出数据写到iceberg样本表 训练框架读iceberg样本表可online training或者batch training 4.2 基于Iceberg MOR的增量特征回填优化 如3.4节的BackFill功能将全量Shitu和Delta Snapshot拼接之后再进行样本计算。这个逻辑存在2个问题一是Shitu数据量比较多拼接效率低。二是每次都需要全量计算所有特征性能开销大。当然可以做增量特征计算在和基线样本拼接。但样本数据量比较大Hive表拼接性能较差在某些情况下可能比全量计算特征慢。
为此我们计划在4.1工作基础上利用iceberg的MOR技术优化BackFill的性能 维护一份基线样本的iceberg表 在基线样本iceberg表新建一个branch增加新特征列 基于Shitu和delta snapshot只做增量特征计算并将增量特征写到新特征列。这一步只计算增量特征不需要join可极大提升性能 训练模块读样本表利用Iceberg MOR的能力读基线特征增量特征再merge成完整的特征列表完成训练 -End-
作者丨lixiaowei、正鼎
