基于 Apache Flink 的实时计算数据流业务引擎在京东零售的实践和落地

摘要：本文整理自京东零售-技术研发与数据中心张颖&闫莉刚在 ApacheCon Asia 2022 的分享。内容主要包括五个方面：

1. 京东零售实时计算的现状

2. 实时计算框架

3. 场景优化：

   TopN

4. 场景优化：动线分析

5. 场景优化：FLINK 一站式机器学习

Tips：点击「阅读原文」查看更多技术内容～

01

京东零售实时计算的现状

1.1 现状

• 技术门槛高、学习成本大、开发周期长。行业内实时开发能力只有少数人能够掌握的现状；
• 数据开发迭代效率比较低，重复逻辑反复的开发缺少复用；
  
• 测试运维难，复杂业务逻辑难以局部测试。
  

1.2 动力

• 降本增效、节省人力，助力高效开发；
• 多角色数据开发，不同角色对应不同的开发方式，非数据人员也能做数据开发的工作。

1.3 目标

• 降低数据开发门槛，通过标准化积木式的开发，实现低代码配置化数据加工，进一步实现图形化清晰表达数据流转；
• 通过算子库组件的沉淀，提升开发效率，提高复用性，一站式加工；
• 通过单元测试以及沉淀用例，提高开发质量。

02

实时计算框架

2.1 为什么做数据流框架

• 数据流框架：9N-Tamias/9N-Combustor，数据流框架基于计算引擎之上，提供一种易用高效的数据开发方式，包括：tamias，是基于 Flink 的引擎的开发框架；combustor：基于 Spark 引擎的开发框架。基于 9N-Tamias 和 9N-Combustor 提供数据流开发工具；
• 支持实时离线统一的表达；
• 多种使用方式：图形化、配置化、SDK 等；
• 算子、组件复用：数据流算子、转换算子、自定义算子、目标源算子，灵活的组合，沉淀常用的算子组合，组件化包括数据流组件和自定义组件，通过数据流开发沉淀数据流组件，同时也开放自主开发自定义组件方式，通过算子、组件的复用，提高开发效率。
  

数据流框架上层各业务场景基于数据流组件化，实现业务数据的加工，包括样本中心、京享值、搜索等一些业务。

2.2 怎么做实时计算框架？

实时计算框架分成四层：

1. Function 层：实现比如 Json 解析、RPC 调用、以及数据流的链接；

2. Process 层：对 Flink 引擎、Data Stream、Data Set、SQL 等 API 进行封装；

3. Function 和 Process 组合生成 Operator，对具体的处理逻辑进行封装，比如实现 Source、Sink、Filter、Join 等常用的算子；

4. 一个或者多个 Operator 构成不同的场景，比如多流拼接导数的 Top N、动线分析，这些构成了 JSON 的配置文件，然后再通过通用的引擎解析配置文件提交任务。

2.3 实时框架：公用 Ops 和 Function

数据接入 Source 和 Sink 层：实现了实时离线、近线常用的数据源；

数据解析 Function：是为了将公用的计算逻辑进一步细化，在算子里封装多个 Function，进行灵活实现业务的逻辑；

算子 Template：如多流拼接、TopN、Count Time Window，业务自己实现会比较复杂，因此框架提供了这些算子的 Template，业务只需要在 Template 的基础上增加业务代码即可，不需要再对这些通用的算子进行学习、开发、调试等工作；

业务算子：可以基于 Template 已有的业务算子，重写得到新的业务算子，也可以自定义组合 Function，形成业务算子。

优点如下：

1. 开发标准化：基于框架提供的公用算子，组合完成业务标准化的开发；

2. 易用性提升：框架提供一些常用且难以实现的算子，使业务的开发变得简单；

3. 开发迭代效率提升：业务只需要关注业务逻辑，从而提高开发迭代效率质量的提升；

4. 质量提升：框架提供的公共算子都是经过严格的测试，并经过长期的业务验证，从而提高开发质量。

03

场景优化：TopN

3.1 复用算子

首先不仅仅是 TopN，包括所有业务场景，数据接入和数据写出都是可以共用的，比如针对流计算，像 Kafka 或 JMQ 的接入和写出，都是可以复用的。

然后是数据解析的算子，包括 JSON 解析、CSV 解析都是可以复用的，但是如果每一个 JSON 解析和 CSV 解析都抽象成一个 Operator，会需要很多的 Operator，因此抽象了 Function 概念，然后 Function 可以组合成公用的算子。

【案例】以榜单计算为例，首先用订单榜单的一个元素值作为一个计算，然后 KeyBy 时用榜单 ID 加元素，接下来再进行一次订单榜单元素值的计算，把榜单 ID 和元素值进行一次 KeyBy，产生的 TopN 的排序。

在这里需要 KeyBy 两次，因为在京东的固有的场景下，有业务上的数据倾斜，只能采用多次聚合，或者是多次排序的方式来解决问题。

3.2 任务优化

HDFS 小文件的问题：因为数据量非常大，因此在写 HDFS 时，如果 Rolling 策略设置不合理，会导致 HDFS 产生很多的小文件，可能会把 HDFS Name Node 的 RPC 请求队列打满。通过源码及其任务机制发现，HDFS 的文件 Rolling 的策略与 Checkpoint 的时间以及 Sink 的并行度相关，因此合理设置 Checkpoint 的时间和 Sink 的并行度，可以有效解决 Sink HDFS 的小文件的问题。

RocksDB 优化：通过查看官方文档可以发现，针对 RocksDB 相关的优化有很多，但是如何有效优化 RocksDB 的设置，核心就在于合理地设置 BlockCache 和 WriteBuffer 的大小，还可以添加 BloomFilter，相应调整这些参数，具体采用哪些配置都可以。

Checkpoint 优化：主要是超时时间、间隔时间、最小停顿时间。比如超时时间是半个小时，这个任务产生了 Fail 了，假如它是在 29 分钟的时候，进行 Failover 的时候，需要从上个 Checkpoint 开始恢复，需要很快消费前 29 分钟的数据。这种情况下如果数据量非常大，对任务是一个不小的冲击。但是如果把 Checkpoint 的时间设置为更合适的 5 分钟或者 10 分钟，这个冲击量会少很多。

数据倾斜：造成数据的倾斜的情况有很多种，比较难解决的是数据源中引发的数据倾斜问题，因此可以采用多次聚合或者多次排序模式解决；另外一个是机器问题，是由于某台机器问题造成的数据倾斜，通常的表现是这台机器上所有的 Subtask 或者 TM 都会产生问题。

04

场景优化：动线分析

4.1 什么是动线

用户点击以及页面展现的浏览路径称之为是动线；以搜索词举例，在京东平台首先搜索台灯，然后又搜索台灯学习，最后搜索儿童学习护眼台灯，从台灯到台灯学习，到儿童学习护眼台灯，这样搜索词的线称为搜索词动线。

动线分析的作用：寻找决定转化的关键路径点以理解用户决策习惯；经常相邻查询的搜索词通过导流工具串联，发现趋势动线；同一个用户对不同排序策略的接受程度，最终从细分的用户类型，提出个性化的导购布局和策略建议；

4.2 数据建模

涉及到串联相邻的搜索词问题，需要从宏观的角度进行数据建模。

首先在京东每天 PB 数据量的动线数据分析下，现有的图结构是没有办法解决这个问题。目前最常用的一个分析方法，是把大批量的这种数据全部同时灌到数据库里，然后等离线数据运行一段时间，拿到分析的结果从结果上去分析。

当前业界在线图数据库进行这种大数据量的图分析，会严重地影响数据库的运行和对外提供服务，因此引入 Flink Gelly 技术栈，通过类似 MySQL 与 Hive 的模式，解决这种大规模图分析问题。

解决方案：首先是把图的源数据通过 Flink SQL 从 Hive 里取出数据，通过 Left Join 把每个 Session ID 下面的 Query 链连起来，然后导入到 HDFS 里；从 HDFS 里读动线的数据，并且把动线的数据生成一个 Graph，根据数据科学家提出的分析条件，将图的分析的结果，直接灌到 OLAP 里进行多维的分析；数据流实时计算的框架，从 Hive 或者 HDFS 里读数据，然后通过数据的 Join，包括写 HDFS、Graph Generate、Graph Analyse 等以可配置化的形式，生成公用算子放到算子库里，对于搜索、推荐或者是广告等所有涉及到动线分析的部门，都可以用到。

4.3 模型建模

如果要对用户进行细分和个性化的分析，就涉及到模型建模。

首先是样本生产的过程，需要把数据从 Hive 里拿到，针对搜索词动线分析需要拿到用户搜索词的表，然后和相应的订单表里决定下单的 Query 进行左连接，生成样本放到 HDFS 里。

训练任务是从 HDFS 里把这些数据灌到 Alink 里进行 Shaply Value 建模，最终的 Query 重要度写到 Hive 里。

全链路是以公用算子的方式提供，目前京东采用这种离线训练的方式，相当于是天级，之后希望天级训练的模式实时化，做成分钟级的或者流式的 Join。

05

场景优化：FLINK 一站式机器学习

机器学习可以从四个方面来描述：特征、样本、训练、预估，而每个方面都有相应的问题（如上图）。

5.1 特征

从生成的角度，特征分为实时特征和离线特征；从特征的特性分为静态特征和动态特征。

1. 静态特征是相对变化不太大的特征，比如用户的年龄、店铺评分、商品金额，可以把静态特征和离线特征相对应；

2. 动态特征比如近一个小时内的点赞量，或者近一个小时内的点击量，动态特征和实时特征相对应。

离线特征可以分为特征的整体生成过程。

1. 特征一般是放到 Hive 里，会涉及到一些特征的解析以及计算，最终生成一个特征的大宽表，然后把这些特征放到 Redis 里，如果是实时特征，涉及到数据接入以及数据解析行为。

2. 特征生成可以认为是业务化的过程，特征写入可以直接写入 Redis 里。

3. FeatureOPS 主要是专注于特征生成，如果特征解析涉及到业务算子，也可以用 FeatureOPS 来做。
5.2 样本

样本分为实时样本拼接和离线样本拼接两个链路；针对样本的特性，有离线的样本和实时的样本两个链路。

离线的样本拼接：通过 Join 存到数仓里，从数仓里拿取用户的曝光以及行为日志后，通过一系列的 Join 操作，形成样本的宽表，每个业务可以从样本宽表拿到属于自己的样本进行模型的训练。

实时的链路拼接也是相同的，区别是样本拼接为实时的。Flink 样本基本上都是双流的，采用 Unit 和 Timer 模式，适配多流的样本拼接，会涉及到大状态的优化，大状态目前用的 State Backend 是 Roll SDB。Watermark 更新机制是采用最慢的时间作为更新的机制，如果某一个行为流的数据量比较少，则会导致 Watermark 不更新的问题。

实时样本拼接针相对离线的样本拼接更加困难，包括一个窗口的选择、一些业务上的样本拼接等。

Sample OPS 做样本质量的校验：首先在样本生成的阶段，需要做样本的分布，如正负样本的分布；其次在做实时样本或者是离线样本拼接时，需要对拼接率做监测；观察任务的延时率，即每一条样本的延时情况。

模型升级定义为只有模型进行模型校正时，才会认为它升级了，而增量训练不是模型升级。

5.3 模型 online learning

模型 online learning 是指数据科学方向，并非大模型的方向。按照特征和样本实时离线的 Template，把模型分为实时和离线两种。

实时训练涉及到模型实时参数的更新，但并非每一条数据训练一次，由超时时间 CountWindow 解决这个问题，比如 Count 达到 1 万条或者超时时间 5 分钟，来解决 Mini Batch 的问题。

针对 Online Learning，目前没有办法离线地做 AB，因此当一批数据进来时，可以先训练出一个模型，同样用这一批数据做 AB，以达到训练和 AB 的一体化。同时用离线的大数据量训练出来的模型，去及时校正实时训练出来的模型，防止模型训偏了；然后任务内部采用 Keyby 方式实现数据并行，解决模型分布式的问题。

举例，如 Profit 模型，是采用报警维度指标来设置，同时在模型产出时将模型推到模型库，然后 Parameter Server 会不停地在模型库里面把当前的模型的参数快照打到模型库里。

5.4 预估

Flink 做预估目前有两种方案：

方案 A 是将模型如 Tensorflow 或者 PyTorch 模型，通过 RPC 的方式或者 HTTP 的方式部署 Server，由 Flink Task 去远程 Invoke RPC 或者 HTTP，会有网络的开销。因为 Flink Task 可能是实时的，也有可能是离线的，所以在 invoke RPC 时，不可能让它随着 Flink 任务的启动而启动，或者随着 Flink 任务的停止而停止，需要有人来运维该 Server。

方案 B 是将模型 Load 到 Flink TM 内部，即在 Flink TM 内部 Inference 该模型，其优点是不用去维护 RPC 或者 HTTP 的 Server，从资源的角度减少了网络开销，节省了资源。

往期精选

▼ 关注「Apache Flink」，获取更多技术干货 ▼

   点击「阅读原文」，查看更多技术内容～