京东搜索排序在线学习的 Flink 优化实践-阿里云开发者社区

开发者社区> 阿里云实时计算Flink> 正文

京东搜索排序在线学习的 Flink 优化实践

简介: 本文由京东搜索算法架构团队分享,主要介绍 Apache Flink 在京东商品搜索排序在线学习中的应用实践

本文由京东搜索算法架构团队分享,主要介绍 Apache Flink 在京东商品搜索排序在线学习中的应用实践。文章的主要大纲如下:

1、背景
2、京东搜索在线学习架构
3、实时样本生成
4、Flink Online Learning
5、监控系统
6、规划总结

一、背景

在京东的商品搜索排序中,经常会遇到搜索结果多样性不足导致系统非最优解的问题。为了解决数据马太效应带来的模型商品排序多样性的不足,我们利用基于二项式汤普森采样建模,但是该算法仍存在对所有用户采用一致的策略,未有效考虑用户和商品的个性化信息。基于该现状,我们采取在线学习,使深度学习和汤普森采样融合,实现个性化多样性排序方案,实时更新模型的关参数。

在该方案中,Flink 主要应用于实时样本的生成和 online learning 的实现。在在线学习过程中,样本是模型训练的基石,在超大规模样本数据的处理上,我们对比了 Flink、Storm 和 Spark Streaming 之后,最终选择用 Flink 作为实时样本流数据的生产以及迭代 online learning 参数的框架。在线学习的整体链路特别长,涉及在线端特征日志、流式特征处理、流式特征与用户行为标签关联、异常样本处理、模型动态参数实时训练与更新等环节,online learning 对样本处理和参数状态处理的准确性和稳定性要求较高,任何一个阶段都有可能出现问题,为此我们接入京东的 observer 体系,拥有完整的全链路监控系统,保证各个阶段数据的稳定性和完整性;下面我们首先介绍一下京东搜索在线学习架构。

二、京东搜索在线学习架构

京东搜索的排序模型系统架构主要包括以下几个部分:

1、Predictor 是模型预估服务,在 load 模型中分为 static 部分和 dynamic 部分,static 部分由离线数据训练得到,主要学习 user 和 doc 的稠密特征表示,dynamic 部分主要包含 doc 粒度的权重向量,这部分由实时的 online learning 任务实时更新。
2、Rank 主要包括一些排序策略,在排序最终结果确定之后,会实时落特征日志,将 doc 的特征按顺序写入特征数据流,作为后续实时样本的数据源(feature)。
3、Feature Collector 的任务是承接在线预估系统发出的特征数据,对下游屏蔽缓存、去重、筛选等在线系统特有逻辑,产出 Query+Doc 粒度的特征流。
4、Sample join 的任务将上面的 feature 数据、曝光、点击、加购、下单等用户行为标签数据作为数据源,通过 Flink 的 union + timer 数据模型关联成为符合业务要求的样本数据,算法可根据目标需求选择不同的标签作为正负样本标记。
5、Online learning 任务负责消费上游生成的实时样本做训练,负责更新 model 的 dynamic 部分。

image.png

三、实时样本生成

Online Learning 对于在线样本生成的时效性和准确性都有很高的要求,同时也对作业的稳定性有很高的要求。在海量用户日志数据实时涌入的情况下,我们不仅要保证作业的数据延时低、样本关联率高且任务稳定,而且作业的吞吐不受影响、资源使用率达到最高。

京东搜索排序在线样本的主要流程如下:

image.png

1、数据源大致有曝光流、feature 流和用户行为流等作为实时样本的数据源,统一以 JDQ 管道流的形式,由京东实时计算平台提供平台支撑。
2、接到 feature 流和曝光流、label 流后,进行数据清洗,得到任务需要的数据格式。
3、拿到各个标准流后,对各个流进行 union 操作,之后进行 keyby。
4、我们在 process function 里面添加 Flink timer 定时器,作为样本生成的实时窗口。
5、将生成的样本实时落入 jdq 和 HDFS,jdq 可以用作后面的 online learning 的 input,HDFS 持久存储样本数据,用于离线训练、增量学习和数据分析。

在线样本任务优化实践:

京东搜索样本数据吞吐量每秒达到 GB 规模,对分布式处理分片、超大状态和异常处理提出很高的优化要求。

1、数据倾斜

使用 keyby 的时候,难免会有数据倾斜的情况,这里我们假设 key 设计合理、 shuffle 方式选择正确、任务没有反压且资源足够使用,由于任务 parallelism 设置导致的数据倾斜的情况。我们先看 Flink 里面 key 是如何被分发到 subtask 上面的。

keygroup = assignToKeyGroup(key, maxParallelism)
subtaskNum = computeOperatorIndexForKeyGroup(maxParallelism, parallelism, keyGroupId)

image.png

假设我们的并发设置的是 300,那么 maxParallelism 就是 512,如此设计,必然导致有的 subtask 分布 1 个 keygroup 有的分配两个,同时也导致了数据自然倾斜。针对上述问题,有两个解决方案:

● 设置并行度为 2 的 n 次方;
● 设置最大并行度为 并行度的 n 倍。

如果使用方案 1 ,调整并发的话只能调整 2 的幂次,建议使用方案 2,且假如 parallelism 为 300,maxParallelism 设置为 1200 的情况下假如数据还是有倾斜,可以再相应的把 maxParallelism 设置大一些保证每个 keygroup 的 key 少一些,如此也可以降低数据倾斜的发生。

2、large checkpoint

在线样本用到了 Flink 的 state,我们之前默认将 state 放到了内存里面,但是随着放量的增加,state 数据量激增,发现 GC 时间特别长,之后改变策略,将 state 放入了 RocksDB,GC 问题得以解决。我们针对 checkpoint 做了如下配置:

● 开启增量 checkpoint;
● 合理设置 checkpoint 的超时时间、间隔时间和最小暂停时间。

image.png

● 让 Flink 自己管理 RocksDB 占用的内存,对 RocksDB 的 blockcache、writebuffer 等进行调优。
● 优化 state 的数据使用,将 state 数据放入多个 state object 里面使用,降低序列化/反序列化的代价。

在任务调优的时候我们发现我们的任务访问 RocksDB 的时间非常长,查看 jstack 发现,很多线程都在等待数据的序列化和反序列化,随着算法特征的逐渐增多,样本中的特征个数超过 500 个,使得每条数据的量级越来越大。但是在做样本关联的时候其实是不需要特征关联的,只需要相应的主键关联就可以了,因此,我们用 ValueState 存储主键,用 MapState/ListState 存储特征等值。当然了还可以将这些特征值存储到外部存储里面,这里就需要对网络 io 和 本地 io 之间的选择做一个取舍。

● failure recovery 的时候开启本地恢复。

由于我们的 checkpoint 数据达到了 TB 级别,一旦任务发生 failover,不管是针对 HDFS 还是针对任务本身,压力都非常大,因此,我们优先使用本地进行 recovery,这样,不仅可以降低 HDFS 的压力还可以增加 recovery 的速度。

四、Flink Online Learning

对于 online learning,我们先介绍一下伯努利汤普森采样算法,假设每个商品的 reward 概率服从 Beta 分布,因此给每个商品维护两个参数成功次数 si 及失败次数 fi,及所有商品的公共先验参数成功次数 α 和失败次数 β。

每次根据商品相应的 Beta 分布采样为最优商品的期望 reward: Q(at) = θi,并选择期望 reward 最大的商品展现给用户。最后根据环境给出真实 reward,更新模型相应的参数达到 online learning 的效果。该参数代表一个商品特征,用一个 n 维向量表示,该向量由原始特征通过 MLP 网络预测得到。原始特征经过 DNN 网络得到一个 N 维向量作为该商品的个性化表征,采用 Logistic Regression 函数建模似然函数,利用 Flink 构建该表征和实时反馈所组成的实时样本,用于不断迭代近似更新参数分布。

image.png

1、数据有序性保证

从 jdq 接过实时样本之后,由于之前并没有保证数据的有序性,这里采用 watermark 机制保证数据的有序性。

2、样本数据处理

把只曝光无行为的商品看做负样本,有点击及后续行为的商品看做正样本,当窗口将达到一定正负比例或数据量时进行一次 batch 训练,迭代出新的参数向量,将商品 embedding 数据放到 Flink 的 state 里面,之后作为 model 的 dynamic 部分更新参数。

3、 同步迭代、异步迭代

个性化 ee 参数在线学习采用异步更新方式的时候,存在参数更新顺序错乱问题,这会降低在线学习模型收敛速度,从而造成了流量的浪费,因此,参数异步更新方式更改为同步更新方式,避免参数读写错乱问题。在同步更新的方式下,存储在 status 中的参数向量需要在下一次训练迭代时使用,若参数发生丢失会使该商品的迭代过程中断,为防止系统风险造成参数丢失,设计了参数双重保障。一般的任务异常或重启后参数可从 checkpoint 或 savepoint 中恢复,如果意外情况下参数无法恢复,从远程在线服务中取回上一版参数并记录到 state。

4、多试验版本支持

在线学习任务使用同一个 Flink 任务来支持多个版本模型在不同实验桶下进行 AB 实验,通过版本号区分不同的 AB 流量桶,对应的实时样本以 docid+version 作为 key 进行处理,迭代过程互不影响。

5、custom serialization

为了提高带宽利用率以及性能的需求,我们内部采用 pb 格式传输数据,经过调研,pb 的传输格式优于 Flink 的兜底的 general class 的 kryo 序列化方式,因此我们采用了 Flink 的 custom serialization 解决方案,直接用 pb 格式在 op 之间传输数据。

五、监控系统

这里我们区分业务全链路监控和任务稳定性相关监控,具体情况下面将详细介绍。

1、全链路监控

整个系统使用京东内部的 observer 平台来实现业务全链路监控,主要包括 predictor 服务相关的监控、feature dump 的 QPS 监控、特征和标签质量监控、关联情况监控、train 相关的监控以及 AB 指标相关的一些监控,如下:

image.png

2、任务稳定性监控

任务稳定性监控这里主要是指 Flink 的任务稳定性监控,链路吞吐量达 GB/s规模,特征消息 QPS 达 10W 规模,且 online learning 的不可间断性,不管对于在线样本任务还是 online learning 的任务,相关监控报警都是必不可少的。

image.png

■ 容器的内存、cpu 监控、thread 个数,gc 监控

image.png

■ 样本相关业务监控

image.png

六、规划总结

Flink 在实时数据处理方面有优秀的性能、容灾、吞吐等表现、算子丰富易上手使用、自然支持批流一体化,且目前已有在线学习的框架开源,做在线学习是个不二的选择,随着机器学习数据规模的扩大和对数据时效性、模型时效性要求的提升,在线学习不仅仅作为离线模型训练的补充,更成为模型系统效率发展的趋势。为此我们做的规划如下:

image.png

作者致谢:感谢实时计算研发部、搜索排序算法团队的支持。

版权声明:本文内容由阿里云实名注册用户自发贡献,版权归原作者所有,阿里云开发者社区不拥有其著作权,亦不承担相应法律责任。具体规则请查看《阿里云开发者社区用户服务协议》和《阿里云开发者社区知识产权保护指引》。如果您发现本社区中有涉嫌抄袭的内容,填写侵权投诉表单进行举报,一经查实,本社区将立刻删除涉嫌侵权内容。

分享:

一套基于Apache Flink构建的一站式、高性能实时大数据处理平台,广泛适用于流式数据处理、离线数据处理、DataLake计算等场景。

官方博客
链接