bilibili基于 Flink 的机器学习工作流平台在 b 站的应用

Posted 2021-06-26 阿里云开发者

简介：介绍 b 站的机器学习工作流平台 ultron 在 b 站多个机器学习场景上的应用。

分享嘉宾：张杨，B 站资深开发工程师

导读：整个机器学习的过程，从数据上报、到特征计算、到模型训练、再到线上部署、最终效果评估，整个流程非常冗长。在 b 站，多个团队都会搭建自己的机器学习链路，来完成各自的机器学习需求，工程效率和数据质量都难以保证。于是我们基于 Flink 社区的 aiflow 项目，构建了整套机器学习的标准工作流平台，加速机器学习流程构建，提升多个场景的数据实效和准确性。本次分享将介绍 b 站的机器学习工作流平台 ultron 在 b 站多个机器学习场景上的应用。

目录：

1、机器学习实时化

2、Flink 在 B 站机器学习的使用

3、机器学习工作流平台构建

4、未来规划

GitHub 地址
https://github.com/apache/flink
欢迎大家给 Flink 点赞送 star~

一、机器学习实时化

首先讲下机器学习的实时化，主要是分为三部分：

• 第一是样本的实时化。传统的机器学习，样本全部都是 t+1，也就是说，今天模型用的是昨天的训练数据，每天早上使用昨天的全天数据训练一次模型；
• 第二是特征的实时化。以前的特征也基本都是 t+1，这样就会带来一些推荐不准确的问题。比如，今天我看了很多新的视频，但给我推荐的却还是一些昨天或者更久之前看到的内容；
• 第三就是模型训练的实时化。我们有了样本的实时化和特征的实时化之后，模型训练也是完全可以做到在线训练实时化的，能带来更实时的推荐效果。

传统离线链路

上图是传统的离线链路图，首先是 APP 产生日志或者服务端产生 log，整个数据会通过数据管道落到 HDFS 上，然后每天 t+1 做一些特征生成和模型训练，特征生成会放到特征存储里面，可能是 redis 或者一些其他的 kv 存储，再给到上面的 inference 在线服务。

传统离线链路的不足

那它有什么问题呢？

• 第一是 t+1 数据模型的特征时效性都很低，很难做到特别高时效性的更新；
• 第二是整个模型训练或者一些特征生产的过程中，每天都要用天级的数据，整个训练或者特征生产的时间非常长，对集群的算力要求非常高。

实时链路

上图我们进行优化之后整个实时链路的过程，红叉的部分是被去掉的。整个数据上报后通过 pipeline 直接落到实时的 kafka，之后会做一个实时特征的生成，还有实时样本的生成，特征结果会写到 feature store 里面去，样本的生成也需要从 feature store 里面去读取一些特征。

生成完样本之后我们直接进行实时训练。整个右边的那个很长的链路已经去掉了，但是离线特征的部分我们还是保存了。因为针对一些特殊特征我们还是要做一些离线计算，比如一些特别复杂不好实时化的或者没有实时化需求的。

二、Flink 在 b 站机器学习的使用

下面讲下我们是怎么做到实时样本、实时特征和实时效果评估的。

• 第一个是实时样本。Flink 目前托管 b 站所有推荐业务样本数据生产流程；
• 第二个是实时特征。目前相当一部分特征都使用了 Flink 进行实时计算，时效性非常高。有很多特征是使用离线 + 实时组合的方式得出结果，历史数据用离线算，实时数据用 Flink，读取特征的时候就用拼接。

  但是，这两套计算逻辑有的时候不能复用，所以我们也在尝试使用 Flink 做批流一体，将特征的定义全部用 Flink 来做，根据业务需要，实时算或者离线算，底层的计算引擎全部是 Flink；

• 第三是实时效果的一个评估，我们使用了 Flink+olap 来打通整个实时计算 + 实时分析链路，进行最终的模型效果评估。

实时样本生成

上图是目前实时样本的生成，是针对整个推荐业务链路的。日志数据落入 kafka 后，首先我们做一个 Flink 的 label-join，把点击和展现进行拼接。结果继续落入 kafka 后，再接一个 Flink 任务进行特征 join，特征 join 会拼接多个特征，有些特征是公域特征，有些是业务方的私域特征。特征的来源比较多样，有离线也有实时。特征全部补全之后，就会生成一个 instance 样本数据落到 kafka，给后面的训练模型使用。

实时特征生成

上图是实时特征的生成，这边列的是一个比较复杂的特征的过程，整个计算流程涉及到了 5 个任务。第一个任务是离线任务，后面有 4 个 Flink 任务，一系列复杂计算后生成的一个特征落到 kafka 里面，再写入 feature-store，然后被在线预测或者实时训练所用到。

实时效果评估

上图是实时效果的评估，推荐算法关注的一个非常核心的指标就是 ctr 点击率，做完 label-join 之后，就可以算出 ctr 数据了，除了进行下一步的样本生成之外，同时会导一份数据到 clickhouse 里面，报表系统对接后就可以看到非常实时的效果。数据本身会带上实验标签，在 clickhouse 里面可以根据标签进行实验区分，看出对应的实验效果。

三、机器学习工作流平台构建

痛点

• 机器学习的整个链路里面有样本生成、特征生成、训练、预测、效果评估，每个部分都要配置开发很多任务，一个模型的上线最终需要横跨多个任务，链路非常长。
• 新的算法同学很难去理解这个复杂链路的全貌，学习成本极高。
• 整个链路的改动牵一发而动全身，非常容易出故障。
• 计算层用到多个引擎，批流混用，语义很难保持一致，同样的逻辑要开发两套，保持没有 gap 也很困难。
• 整个实时化成本门槛也比较高，需要有很强的实时离线能力，很多小的业务团队在没有平台支持下难以完成。

上图是一个模型从数据准备到训练的大概过程，中间涉及到了七八个节点，那我们能不能在一个平台上完成所有的流程操作？我们为什么要用 Flink？是因为我们团队实时计算平台是基于 Flink 来做的，我们也看到了 Flink 在批流一体上的潜力以及在实时模型训练和部署上一些未来发展路径。

引入 Aiflow

Aiflow 是阿里的 Flink 生态团队开源的一套机器学习工作流平台，专注于流程和整个机器学习链路的标准化。去年八、九月份，我们在和他们接触后，引入了这样一套系统，一起共建完善，并开始逐渐在 b 站落地。它把整个机器学习抽象成图上的 example、transform 、Train、validation、inference 这些过程。在项目架构上非常核心的能力调度就是支持流批混合依赖，元数据层支持模型管理，非常方便的进行模型的迭代更新。我们基于此搭建了我们的机器学习工作流平台。

平台特性

接下来讲一下平台特性：

• 第一是使用 Python 定义工作流。在 ai 方向，大家用 Python 还是比较多的，我们也参考了一些外部的，像 Netflix 也是使用 Python 来定义这种机器学习的工作流。
• 第二是支持批流任务混合依赖。在一个完整链路里面，涉及到的实时离线过程都可以加入到里面，并且批流任务之间可以通过信号就行互相依赖。
• 第三是支持一键克隆整个实验过程。从原始 log 到最终整个实验拉起训练这块，我们是希望能够一键整体链路克隆，快速拉起一个全新的实验链路。
• 第四是一些性能方面的优化，支持资源共享。
• 第五是支持特征回溯批流一体。很多特征的冷启动需要计算历史很长时间的数据，专门为冷启动写一套离线特征计算逻辑成本非常高，而且很难和实时特征计算结果对齐，我们支持直接在实时链路上来回溯离线特征。

基本架构

上图是基本架构，最上面是业务，最下面是引擎。目前支持的引擎也比较多：Flink、spark、Hive、kafka、Hbase、Redis。其中有计算引擎，也有存储引擎。以 aiflow 作为中间的工作流程管理，Flink 作为核心的计算引擎，来设计整个工流平台。

工作流描述

整个工作流是用 Python 来描述的，在 python 里面用户只需要定义计算节点和资源节点，以及这些节点之间的依赖关系即可，语法有点像调度框架 airflow。

依赖关系定义

批流的依赖关系主要有 4 种：流到批，流到流，批到流，批到批。基本可以满足目前我们业务上的所有需求。

资源共享

资源共享主要是用来做性能方面，因为很多时候一个机器的学习链路非常长，比如刚刚那个图里面我经常改动的可能只有五六个节点，当我想重新拉起整个实验流程，把整个图克隆一遍，中间我只需要改动其中的部分节点或者大部分节点，上游节点是可以做数据共享的。

这个是技术上的实现，克隆之后对共享节点做了一个状态追踪。

实时训练

上图是实时训练的过程。特征穿越是一个非常常见的问题，多个计算任务的进度不一致时就会发生。在工作流平台里面，我们定义好各个节点的依赖关系即可，一旦节点之间发生了依赖，处理进度就会进行同步，通俗来说就是快的等慢的，避免特征穿越。在 Flink 里面我们是使用 watermark 来定义处理进度。

特征回溯

上图是特征回溯的过程，我们使用实时链路，直接去回溯它历史数据。离线和实时数据毕竟不同，这中间有很多问题需要解决，因此也用到了 spark，后面这块我们会改成 Flink。

特征回溯的问题

特征回溯有几个比较大的问题：

• 第一是如何保证数据的顺序性。实时数据有个隐含的语义就是数据是顺序进来的，生产出来立马处理，天然有一定的顺序性。但是离线的 HDFS 不是，HDFS 是有分区的，分区内的数据完全乱序，实际业务里面大量计算过程是依赖时序的，如何解决离线数据的乱序是一个很大的问题。
• 第二是如何保证特征和样本版本的一致性。比如有两条链路，一条是特征的生产，一条是样本生产，样本生产依赖特征生产，如何保证它们之间版本的一致性，没有穿越？
• 第三就是如何保证实时链路和回溯链路计算逻辑的一致？这个问题其实对我们来说不用担心，我们是直接在实时链路上回溯离线数据。
• 第四是一些性能方面的问题，怎么快速得算完大量的历史数据。

解决方案

以下是第一、第二个问题的解决方案：

• 第一个问题。为了数据的顺序性，我们 HDFS 的离线数据进行 kafka 化处理，这里不是把它灌到 kafka 里面去，而是模拟 kafka 的数据架构，分区并且分区内有序，我们把 HDFS 数据也处理成类似的架构，模拟成逻辑上的分区，并且逻辑分区内有序，Flink 读取的 hdfssource 也进行了对应的开发支持这种模拟的数据架构。这块的模拟计算目前是使用 spark 做的，后面我们会改成 Flink。

• 第二个问题分为两部分：

  • 实时特征部分的解决依赖于 Hbase 存储，Hbase 支持根据版本查询。特征计算完后直接按照版本写入 Hbase，样本生成的时候去查 Hbase 带上对应的版本号即可，这里面的版本通常是数据时间。
  • 离线特征部分，因为不需要重新计算了，离线存储 hdfs 都有，但是不支持点查，这块进行 kv 化处理就好，为了性能我们做了异步预加载。

异步预加载的过程如图。

四、未来规划

接下来介绍下我们后面规划。

• 一个是数据质量保证。现在整个链路越来越长，可能有 10 个节点、 20 个节点，那怎么在整个链路出问题的时候快速发现问题点。这里我们是想针对节点集来做 dpc，对每个节点我们可以自定义一些数据质量校验规则，数据通过旁路到统一的 dqc-center 进行规则运算告警。

• 第二是全链路的 exactly once，工作流节点之间如何保证精确一致，这块目前还没有想清楚。

• 第三是我们会在工作流里面加入模型训练和部署的节点。训练和部署可以是连接到别的平台，也可能是 Flink 本身支持的训练模型和部署服务。

嘉宾介绍：张杨，17 年入职 b 站，从事大数据方面工作。

原文链接：https://developer.aliyun.com/article/784356?

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