一、京东物流业务介绍

null

京东集团在 2007 年开始自建物流，是国内领先的以技术驱动的供应链解决方案及物流供应商，一体化的供应链物流是我们的核心赛道。

二、物流风控场景概括

null

京东物流风控场景主要概括为两种，一种是货品安全，如货物丢失、破损等。另一种是交易风险，主要包括财务支出的一些风控场景。这两种场景主要体现在销售、仓储、运力、计重和终端。最后的防损风控属于事后风控，任何有防损诉求的风控都会被归纳到防损风控中。

其中销售风控的主体是商家。从准入、运营、财务、理赔各个环节进行风险管控。为了解决长期以来受不良客户扰乱市场的困扰，我们整合内部资源，通过风险商家模型和企业、法人知识图谱的搭建，对风险行为进行精准识别、分析、预警及处置。

仓储风控的主体是 sku。从单一人工举报发现渠道，增加了利用数据发现渠道，扩充了发现风险线索的范围，持续输出风险线索，支持运营人员日常发现风险，及时核查并规避风险。

运力风控的主体是车辆。通过事前供应商招采，事中运营结算，事后防损稽核的方式，将已经完结的任务中可能存在的风险进行识别，及时扣减不合理的费用，降低公司的损失。

计重风控主要是指运单称重量方。通过预定的规则，针对特定的运单重新进行称重量方。这么做是为了避免某些主观上的偷重漏重，或者因为作业规范、设备等问题导致的收入损失。

终端风控防控的对象是人。通过预测画像优化业务流程，从下单到配送针对性的进行风险规避。

防损风控是利用大数据分析、算法策略构建的大量损耗模型，并通过调查预审、案件平台等业务系统识别出应收、丢损、异常等风险信息形成案件。督导防损调查人员及时调查处置。

三、物流风控平台建设

null

整个风控技术架构主要包含来源层、采集层、处理层、存储层、应用层。数据来源于业务数据、财务数据、物联网数据、外部数据。通过批、流、数据导入的方式，经过数据挖掘、机器学习、搜索引擎、算法服务，最终对外提供服务，服务包括数据服务、特征服务、标签服务、画像服务。

null

在进行具体的风控平台搭建之前，我们先了解一下风控的定义。风险控制存在四个基本方法，风险回避、损失控制、风险转移、风险保留。

京东物流主要集中在损失控制上，包含事前、事中、事后三个方面。

事前：预防性风控，通过外部数据、历史规律、算法预测来挖掘出现风险的概率，并按照预定的预防措施和保护性措施，降低损失的概率。
事中：阻断性风控，主要是在流程进行中，系统根据相关规则监控，在出现实际损失前启动拦截流程，降低损失的金额。
事后：处置性风控，主要围绕风险处置及复盘来做的，属于挽损性质。

null

了解了风控的定义，我们再来分享一下平台的建设思路。

第一，中台服务，对外提供能力，尽量不直接操作生产流程。过多的嵌入到生产流程，会影响到生产的效率。
第二，风控平台接收各方数据源，对各场景进行风险识别、打标并落库，建立风险库。风控模型的搭建是一个长期的不断迭代优化的过程。
第三，风控平台提供可配置的风控识别能力，用户可以自己选择风险识别类型，根据自身需求配置参数。这样做是为了扩大风控模型的应用范围，提升投入产出比。
第四，风控平台提供标准化接入方式，无论消息还是查询接口，只要满足需求，都可以接入并使用。一个成熟的平台必然是标准化的，通过接入流程化，风控平台的技术人员只需专注于模型的识别效率提升即可。
第五，风控平台尽量少的根据特定场景进行逻辑加工，不直接嵌入用户的代码。这么做也是为了扩大风控模型的应用范围，做到以点破面。

null

风控数据库包含规则库、模型库、知识图谱，运用的数据库类型也是市面上常见的关系数据库、NoSQL 数据库、搜索引擎、图数据库等。现行的方案是 Flink 实时更新，实时模型和知识图谱，Flink SQL 或者 Hive SQL 更新离线模型，同时兼顾数据回算。

从上图可以看出实时和离线是相互独立的。左上部分是实时风控，它包含事中和事后风控流程，主要根据生产活动所产生的日志信息、维度接口、预先设定的规则模型，进行风险指标计算。通过风险预警将风险结果反馈给生产方，同时对接风控大盘供领导决策。

中间部分是事中风控模型，运用的技术有实时也有离线。主要流程是生产系统主动发起是否需要风控、是否黑白名单、根据规则判断是否命中风控。如果命中，生产系统会有相应的管控流程，会将管控的结果同步给财务系统，财务系统在闭环入仓形成一个循环。

最下部分是事后和事前风控流程，主要依据历史数据和过往风险判断的结果，根据一系列大数据手段对风控模型进行迭代优化，从数据中发现问题并形成案件，推送给调查预审、案件平台等运营系统，同时督导防损调查人员进行防损调查。调查的结果，除了反馈给模型，还会同步给财务，从而挽回损失。

null

Flink 作为实时计算引擎，对接两部分数据，分别是业务数据和规则数据。规则应用将变化的规则数据通过消息的形式流入到计算引擎中。当 Flink 接收到实时生产消息时，通过数据清洗、数据压缩、数据封装，和预先流入的规则信息相结合，进行风险指标计算。计算结果保存进实时模型中，并通过消息或者接口的形式反馈给风控预警以及风控大盘。

四、Flink 赋能

null

我们最开始没有事前和事中风控，属于大数据风控，风险的判定比较滞后，严重影响到物流风控工作的推进。之后在我们初步涉及实时风控时，并没有选择实时计算引擎，而是采取 docker+多线程的方式，这就使得各个场景割裂，无法统一管理，且消费速度较慢。在数据洪峰到达时，资源利用率较低、扩容困难；在一些注重时效的事中风控中，命中率较低。

现在物流风控各环节，大规模推进事中风控，原有架构就已经无法满足了。从运力事中风控开始逐步迁往 Flink 流式计算引擎，并开始搭建风控实时中台模型，满足业务场景对高吞吐、低延迟、高性能的要求，同时积极探索流批一体在风控场景下的应用。

null

先来简单了解一下批处理和流处理的概念，流处理是通过 SDK、Storm、Flink 等流式计算引擎对数据进行逐条处理，并将处理结果保存到数据库中。批处理则是通过 Hive、Spark 对数据进行分层，按批处理，从 ODS 到 DWS，最终输出结果对外服务。

null

我们理想的计算引擎应该是统一的，统一的数据源、计算过程和服务。而 Flink 正是未来的统一计算引擎，它有着以下几个特点：

成熟的流批一体概念：统一的 shuffle 架构设计，并专门对序列化和内存拷贝进行了优化。
生态兼容：与 Hadoop yarn /Kubernetes 集成，并且支持单机模式运行。
性能卓越：性能卓越的批处理和流处理支持。
规模计算：作业可被分解成为上千的任务，分布在集群中并发执行。满足我们对高性能、高时效的需求。

五、技术挑战

null

这是计重风控、终端风控相关的部分数据链路，从仓到分拣再到配送，整个数据链路长，风险规则繁多。

null

今年双十一大促时，单个拓扑的消息总量达到百亿级以上，因此我们面临诸多的挑战。

业务复杂性：多业务条线、多生产环节；数据源多，数据异构；数据生命周期长；数据复式波峰；业务迭代快。
海量数据：大促多条线数据量爆发，对集群产生了很大的压力。
大状态：单次增量 state 达 40G，高峰是更是达到 100G 以上。
高实时性：终端风控场景要求时效控制在 3 秒内（包含上下游网络差异）
数据倾斜：数据源上数据分布不均匀。
存储压力：存储无法抗住 Flink 的写入 QPS。

null

我们最先开始的是业务上的优化，对于数据倾斜问题，我们采用了感兴趣列过滤，减少进入拓扑的数据数量，主要依靠 kafka 当前变更的值来进行数据过滤。反压监控，优化逻辑或增加资源。根据业务具体条件，变更 keyby 字段。

对于高实时性的问题，我们采取变更数据源、减少途径链路、前置计算、添加缓存、剥离核心和非核心业务。减少数据链路的同时，尽可能降低拓扑的大小。

对于存储的问题，我们进行压缩合并，降低写入频率。将计算写入分离，避免因为存储问题影响集群的正常运行。

null

因为每一个拓扑的数据量、计算逻辑都不相同，所以每个配置也不尽相同。要进行合理的配置需要了解 TM 的内存模型。从图上所知，如果是容器化部署，那么容器的内存大小就是总处理内存的大小。Flink 总内存的大小等于总处理内存减去元空间和内存超限的大小，这两块一般采用默认值即可。

Flink 总内存包含堆和非堆两部分，框架堆内存和框架非堆内存也是采用默认值就可以，无需额外配置。任务堆内存等于 Flink 总内存减去其他内存配置的大小得出，也无需单独配置。

这么算下来，我们只需关注托管内存、任务堆外内存以及网络缓存即可。托管内存如果使用 RocksDB 状态后端，且状态数据量较大或读写较频繁，建议适当增加托管内存的大小。配置太小对集群性能的影响是非常巨大的，可以配合 RocksDB 监控来决定。

任务堆外内存一般流作业很少用到，所以可以优先保障堆内存，降低该内存的大小。网络缓存也是一般容易造成浪费的地方，整个拓扑需要的网络缓存大小是由并行度以及上下游交互方式所决定，所以在拓扑启动时，就会确定一个大小，且向上浮动不大。我们一般保留 50%~70%来预防后续并行度增加即可，优先保障任务堆内存。

null

先了解下状态的写入和读取流程。写入操作先写入活动写缓存，写满以后转换为只读写缓存，再 flash 进磁盘形成 SST 索引文件。读取则是依次从活动写缓存、只读写缓存、块缓存以及 SST 索引文件中寻找目标数据。因此我们只需关注写缓存、块缓存以及 SST 索引文件这三部分的内存大小即可。

一般情况，这三者的配置无需变更，若作业状态特别重读或重写，可以适当进行调整，但优先保证托管内存的充足。还有一些其他建议调整的参数，比如启用增量状态同步、开启压缩清理、调整状态异步的线程数量，这个一般与容器的 CPU 个数保持一致即可。

状态预定配置有四种，默认、机械磁盘、机械磁盘+内存、SSD。如果单个槽的状态量达到 GB 级别，且托管内存充裕的情况下，设置为机械磁盘+内存的性能是最佳的，其他情况设置为机械磁盘即可。

null

还有一些其他优化，代码层面，使用元组、复用对象、数据去重、异步调用等等。

部署层面上，并行度优化、链合并、拓扑分拆、计算和写入分离、采用 child-first 的类加载机制避免版本冲突等等。

六、未来规划

null

目前流批一体在风控场景上的应用还是比较薄弱的，旧版本的升级也是未来规划中比较重要的一环。风险库的建立还处于野蛮发展中，其中的规范制定也是迫在眉睫的事情。

原文阅读