【干货篇】bilibili：基于 Flink 的机器学习工作流平台在 B 站的应用

分享嘉宾：张杨，B 站资深开发工程师
整理出品： AICUG 人工智能社区

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

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

01 机器学习实时化

首先讲下机器学习的实时化，主要是分为三部分：第一是样本的实时化。传统的机器学习，样本全部都是 t+1，就是今天模型用的是昨天的训练数据，每天早上使用昨天的全天数据训练一次模型。

第二是特征的实时化，以前特征基本全部也是 t+1，这样就会带来一些推荐的问题，很不准，可能今天我看到很多新的视频，但是给我推荐的还是一些我昨天看到或者更久之前的一些东西。

第三就是模型训练的实时化，我们有了样本的实时化和特征的实时化之后，那模型训练也是完全可以做到在线训练实时化的，能带来更实时的推荐效果。

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

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

这是我们进行优化之后整个实时链路的过程，红叉的部分是被去掉的。整个数据上报后通过 pipeline 直接落到实时的 kafka，之后会做一个实时特征的生成，还有实时样本的生成，特征结果的会写到 feature store 里面去，样本的生成也需要从 feature store 里面去读区一些特征。生成完样本之后我们直接进行实时训练。整个右边的一个很长的链路已经去掉了，但是离线特征的部分我们还是保存了。因为一些特殊特征我们还是要做一些离线计算，比如一些特别复杂的不好实时化的或者没有实时化需求的。

02 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 里面可以根据标签进行实验区分，看出对应的实验效果。

03 机器学习工作流平台构建

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

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

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

接下来讲一下平台特性。第一是使用 Python 定义工作流，因为在 ai 方向大家用 Python 还是比较多的，我们也参考了一些外部的，像 Netflix 也是使用 Python 来定义这种机器学习的工作流。

第二是支持批流任务混合依赖。在一个完整链路里面，涉及到的实时离线过程都可以加入到里面，并且批流任务之间可以通过信号就行互相依赖。

本文地址：【干货篇】bilibili：基于 Flink 的机器学习工作流平台在 B 站的应用
本文版权归作者和AIQ共有，欢迎转载，但未经作者同意必须保留此段声明，且在文章页面明显位置给出

第三是支持一键克隆整个实验过程。从原始 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 化处理就好，为了性能我们做了异步预加载。