搭建基于Flink的风控系统阿里巴巴大规模风控技术难点实战目前Flink基本服务于集团所有BU.容量达到每秒40亿条，计算任务达到3万多个，累计使用100万+个Core；涵盖集团内几乎所有具体业务，如：数据中台、AI中台、风控中台、实时运维、搜索推荐等。01基于Flink构建风控体系风控是一个话题很大，涉及规则引擎、NoSQLDB、CEP等。本章主要讲风控的一些基本概念。在大数据侧，我们把风控分成一个3×2的关系：2表示风控要么是基于规则，要么是基于算法，??要么是基于模型；控制和事后风险控制。1.1对于三类风控业务，事中风控和事后风控在终端上的感知是异步的，事前风控在终端上的感知是同步的。这里稍微解释一下事前风控。事前风控是将训练好的模型或者计算好的数据存储在Redis、MongoDB等数据库中；规则引擎直接从Redis和MongoDB中取数据返回结果；另一种方法是基于KubeflowKFserving，在端有请求后，根据训练好的算法和模型返回结果。一般来说，这两种方式的延迟在200毫秒左右，可以作为同步RPC或者HTTP请求。对于Flink相关的大数据场景，属于异步风控请求，其异步时效性很低，一般是一两秒。如果追求超低延迟，可以把它看成是事件中的一种风控，风控决策过程可以由机器来处理。一种很常见的是用FlinkSQL做指标阈值统计，用FlinkCEP做行为序列规则分析，还有一种是用TensorflowonFlink来描述Tensorflow中的算法，然后用Flink来执行Tensorflow规则的计算。1.2Flink是规则风控的最佳选择目前，Flink是阿里集团风控的最佳选择。主要有3个原因：事件驱动的毫秒级延时流批量集成1.3规则风控三要素规则风控有3个要素，后面介绍的内容都是围绕这3个：事实：指的是风控事件，可能来自业务方或日志埋点，是整个风控系统的输入；规则：往往来自于业务方的定义，即这条规则应该满足什么样的业务目标；Threshold阈值：规则对应的描述的严重性。1.4Flink规则表达的增强对于Flink来说，可以分为无状态规则和有状态规则两种。其中Stateful规则是Flink风控的核心：stateless规则：主要针对数据的ETL。一种场景是当某个事件的某个字段值大于X时触发当前风控行为；另一种场景是Flink任务的下游是基于模型或者算法的风控，Flink端不需要做规则判断，只需要对数据进行向量化，归一化，比如多流关联，CaseWhen判断等，将数据变成0/1的向量，然后推送到下游的TensorFlow进行预测。Statefulrules:Statisticalrules:基于统计分析的计算规则。例如5分钟内访问次数大于100次，则认为触发风控；时序规则：在事件序列中，一个事件对前后事件都有影响，比如点击、加入购物车、删除三个事件。这种连续的行为序列是一种特殊行为。可以认为该行为是恶意降低商家商品的评价分数，但这三个事件并不是独立存在的风险。控制事件；阿里云实时计算Flink完善了基于序列的规则能力，为云上和集团内的电商交易场景提供技术支持；混合规则：统计规则和顺序规则的组合。02阿里风控实战本章主要介绍阿里是如何从工程上满足上述风控三要素的。从整体技术来看，目前分为感知、处置和洞察三个模块：并输出此类异常的列表；再如，由于某年骑行政策调整，导致头盔销量增加，相关商品的点击率和转化率增加。这种情况需要及时感知和捕捉，因为这是正常行为，而不是作弊；disposal：即如何执行规则。现在有三道防线：每小时、实时和离线。与之前单一策略的匹配相比，关联融合后的准确率会更高。例如，对最近一段时间的联想，是根据某些用户在网络中的持续行为综合研判；insight：为了找到一些目前没有被感知到的，不能直接用规则描述的风控行为，比如风控需要用高度抽象的方式来表示样本，需要投射到合适的子空间，然后结合时间维度在高维度上寻找一些特征来识别新的异常。2.1Phase1：SQL实时关联&实时统计这个阶段有一个基于SQL的测评风控系统，用简单的SQL做一些实时关联统计，比如用SQL做聚合操作SUM(amount)>50，其中规则为SUM(amount)，则规则对应的阈值为50；假设现在有10、20、50、100这4条规则同时在线运行，因为单个FlinkSQL作业只能执行一条规则，那么需要为这4条规则设置阈值申请4Flink作业分别。优点是开发逻辑简单，作业隔离度高，缺点是极大浪费计算资源。2.2Phase2：BroadcastStreamPhase1风控规则的主要问题是规则和门槛是不可变的。Flink社区目前有一些解决方案，比如基于BroadcastStream。下图中，TransactionSource负责事件的接入。RuleSource是一个BroadcastStream，当有新的threshold时，可以通过BroadcastStream广播给各个operator。比如判断风控对象在一分钟内连续访问超过10次，但是可能要在618或者双11改成20次或者30次才会被线上感知风险控制系统的下游系统。在第一阶段，只有两种选择：第一种是在线运行所有作业；二是在某个时刻停止一个Flink作业，根据一个新的指标开始一个新的作业。如果是基于BroadcastStream，可以实现规则指标阈值的分发，无需重启job，直接修改在线指标阈值。2.3Phase3：DynamicCEPPhase2的主要问题是它只能更新索引阈值。虽然大大方便了业务系统，但实际上很难满足上层业务。诉求主要有两个：结合CEP实现行为序列的感知；结合CEP仍然可以动态修改阈值甚至规则本身。第三阶段，阿里云Flink对CEP相关进行了高度抽象，将CEP规则与CEP执行节点解耦，即规则可以存储在RDS、Hologres等外部第三方存储中，并且CEP作业发布后，可以加载数据库CEP中的CEP规则是用来实现动态替换的，这样作业的表现力会增强。其次，工作的灵活性将得到增强。比如你想看到某个APP下的一些行为，更新这个行为的指标阈值，可以通过第三方存储更新CEP规则，而不是Flink本身。这样做的另一个好处是可以将规则暴露给上层业务方，让业务真正去编写风控规则。我们成为一个真正的规则中心，这是动态CEP能力的好处。在阿里云的服务中，最新版本已经集成了动态CEP能力。阿里云全托管的Flink服务大大简化了风控场景的开发周期。2.4Phase4：SharedComputing在Phase3的基础上又向前迈进了一步，阿里云实现了“共享计算”的解决方案。在这个共享计算方案中，CEP规则可以通过建模平台进行完整的描述，并作为一个非常友好的规则描述平台暴露给上层客户或者业务方，可以通过拖拽或者其他方式进行耦合，以及然后在调度引擎中选择要运行规则的事件传入源。比如现在两个模型都服务于淘宝APP，完全可以落到同一个Fact的FlinkCEP作业中，做到业务端、执行层、引擎层完全解耦。目前，阿里云的共享计算解决方案非常成熟，有丰富的客户实施实践。2.5第五阶段：引擎端、平台端、业务端业务开发和平台建设分离。Phase4可以实现引擎端和平台端的解耦，但是仍然和业务端高度绑定。两者的工作模式仍然是甲乙双方的协作关系，即业务方控制业务规则，平台方接受业务团队的风控需求制定风控规则。但平台团队通常以人员为主，业务团队会随着业务的发展越来越强大。这时候业务方自己可以抽象出一些基本概念，沉淀一些通用的业务规范，组装成友好的DSL，然后通过阿里云全解耦的OpenAPI实现作业提交。由于需要同时支持集团内近百个BU，无法针对每个BU提供定制化支持。我们只能把引擎的能力尽可能开放，然后把业务端通过DSL封装提交给平台，真正实现的时候，只有一个中台暴露给客户。03大规模风控的技术难点本章主要介绍大规模风控的一些技术难点，以及阿里云如何在全托管Flink商业产品中突破这些技术难点。3.1细粒度的资源调整在流计算系统中，数据源往往不是阻塞节点。上游的数据读取节点因为没有计算逻辑所以不存在性能问题，而下游的数据处理节点是整个任务的性能瓶颈。由于Flink的作业是按槽划分资源的，因此默认源节点和工作节点具有相同的并发度。在这种情况下，我们希望源节点和CEPworker节点的并发度可以分别进行调整。例如下图中，我们可以看到一个作业的CEPworker节点的并发可以达到2000，而源节点只需要2个并行度，就可以大大提升CEP节点的性能。另外，它划分了CEP工作节点所在的TM内存和CPU资源。在开源的Flink中，TM整体上是同构的，也就是说源节点和工作节点的规格是完全一样的。从节省资源的角度来看，源节点在实际生产环境中不需要像CEP节点那么多的内存和CPU资源。源节点只需要很小的CPU和内存就可以满足数据抓取。阿里云全托管Flink使得源节点和CEP节点可以运行在异构TM上，即CEP工作节点TM资源明显大于源节点TM资源，CEP工作执行效率会变得更高。考虑到资源细粒度调整带来的优化，云上全托管服务相比自建IDCFlink可以节省20%的成本。3.2流批集成&自适应批调度如果流引擎和批引擎不采用同一套执行模式，往往会遇到数据标准不一致的情况。出现这个问题的原因是流规则很难在批规则下真正完成。形容它;比如Flink中有专门的UDF，而Spark引擎中没有对应的UDF。当这样的数据口径不一致时，选择哪方面的数据口径就成了一个很重要的问题。在Flink流批一体的基础上，流模式中描述的CEP规则可以在批模式下以相同的口径再次运行得到相同的结果，因此无需开发批模式相关的CEP作业。在此之上，阿里实现了一个自适应的BatchScheduler。事实上，CEP规则每天的效果输出不一定是平衡的。比如今天的行为序列没有异常行为，下游只有少量数据输入。这时候会预留一个elasticcluster用于batchanalysis；当CEP的结果很少时，下游批分析只需要非常少的资源，甚至不需要一开始就指定每个批分析工作节点的并行度，工作节点可以根据输出上游数据和任务负载自动调整批处理并行度，真正做到弹性批处理，这是阿里云Flink流批一体BatchScheduler的独特优势。3.3结合阅读，减轻大众层的压力。这是实践中遇到的问题。现在的开发模式基本都是基于数据中心，比如实时数仓。在实时数仓场景下，可能数据源不多，但是中间层DWD会变多，中间层可能会演变成很多DWS层，甚至会演变成很多数据集市供各个部门使用，在这样的话，单表的读取压力会很大。通常多个源表相互关联（拓宽）形成一个DWD层，从单个源表的角度看依赖于多个DWD表。DWD层也被多个不同业务领域的作业消费，形成DWS。基于这种情况，阿里实现了源码归并。只需要读取一次DWD，Flink端会帮你处理成业务领域的多张DWS表，可以大大降低公共层的执行压力。3.4KV分离设计的状态后端CEP节点在执行时会涉及到非常大规模的本地数据读取，尤其是在行为序列的计算模式中，因为需要缓存之前所有的数据或者某段时间的行为序列。在这种情况下，一个比较大的问题是后端状态存储（如：RocksDB）有非常大的性能开销，会影响CEP节点的性能。目前，阿里巴巴已经实现了专为KV分离设计的状态后端。阿里云Flink默认使用Gemini作为状态后端。在CEP场景中测得的性能至少有100%的提升。3.5维度数据分区加载风险控制很多时候，分析都是基于历史行为。历史行为数据一般存储在Hive或者ODPS表中，这个表的规模可能是TB级别。开源的Flink默认需要在每个维表节点上加载这个超大的维表，这其实是不现实的。阿里云对内存数据实现了基于shuffle的分区，维表节点只会加载属于当前shuffle分区的数据。
