您可以捐助,支持我们的公益事业。

1元 10元 50元





认证码:  验证码,看不清楚?请点击刷新验证码 必填



  求知 文章 文库 Lib 视频 iPerson 课程 认证 咨询 工具 讲座 Model Center   Code  
会员   
   
 
     
   
 订阅
  捐助
金融风控数据管理——海量金融数据离线监控方法
 
作者: 程哲豪
  414  次浏览      51
 2021-5-24
 
编辑推荐:
本文详细论述了该方案TaskMaker、 Calculator、 Checker等各个模块的设计实现,最终实现了用户无需额外开发,只需要简单的配置即可完成监控指标的计算和推送,避免了人力浪费,提升效能。
本文来自于腾讯大讲堂,由Alice编辑、推荐。

背景

在金融业务上,质量和稳定是生命线,我们需要对所有已经上线的风控要素,如策略、模型、标签、特征等构建监控。在过去,我们部署监控的方式为:

风控要素负责同学在要素上线前,通过spark\sql完成对监控指标的运算并例行化;

将监控指标运算结果出库mysql\tbase,用于指标的展示和告警;

告警系统轮询指标是否异常,如异常则通过企业微信等推送告警消息。

这种模式主要的问题在于:

开发门槛高,要素负责同学需要掌握spark离线计算、mysql等数据库的增删数据,还需要手动配置例行化任务,在告警系统上登记注册等,耗时费力;

重复工作多,要素指标相似、重合度很高,如多数风控要素都涉及PSI计算,只是告警阈值不一样;指标出库、配置告警等同样是重复相似操作。

为了解决上述问题,我们设计开发了一套“统一监控计算检查工具”(以下简称监控工具),将监控计算拆解成计算任务生成、监控指标计算、监控指标衍生与检查等模块,实现用户无需额外开发,只需要简单的配置即可完成监控指标的计算和推送,避免了人力浪费,提升效能。

整体设计

系统交互视图

与统一监控计算与检查工具交互的主要有接入方和告警系统,所有的监控由接入方发起,接入方可以是特征、标签、模型、策略的兜底检查同学,也可以是具体业务同学。接入方提出监控需求(填写配置),统一监控计算与检查工具根据需求生成计算任务完成计算,如果触发告警则通过告警系统将告警发送给接入方,接入方接受告警后及时修复并反馈登记,监控工具会读取用户的告警反馈重新完成相关计算,直至监控指标在告警阈值内。

为了完成监控指标的计算,统一监控计算与检查工具可以细分为三个核心模块,分别为:

计算任务生成模块 TaskMaker: 根据配置和被监控表调度周期(hour/day/month)、时间偏置生成监控指标计算任务,TaskMaker解决了不同计算任务例行化周期不同的问题,使得下游模块可以专注于计算本身。

监控指标计算模块 Calculator: 读取未完成的计算任务,计算相关监控指标。Calculator通过生成执行计划并优化的方式,合并不同业务同学对同一表的监控计算需求,提升计算效率。

监控指标衍生与检查模块 Checker: 读取监控指标计算结果,进行环比变化率等衍生,然后对衍生结果进行检查,返回检查结果。

需要注意的是,我们提出了“监控指标衍生的概念”,将不依赖数据源表只依赖监控指标及其历史记录的一类指标称为“衍生指标”,将衍生指标延迟到检查器Checker上计算,可以节省大量计算资源。具体来看,非衍生指标和衍生指标的不同在于:

非衍生指标。非衍生指标即指标计算仅仅依赖于数据源表,而不依赖与历史的监控指标,例如PSI值、迁移率等,这些指标描述了监控要素分布的变化,其计算只依赖于源表的当前周期和对比周期数据,不需要对监控指标进行衍生,如PSI>0.1即告警。此外任务或表状态监控,如任务完成时间、表分区计数等也只依赖源表,不需要衍生。非衍生指标只能由Calculator完成计算,通常需要多次遍历数据源表,监控所消耗的主要计算资源就是计算非衍生指标所导致。

衍生指标。衍生指标是指对监控指标进行二次运算后得到的监控指标,衍生指标的计算不依赖源表,只依赖监控指标及其历史记录。例如零值、缺失值率,它们是非衍生指标,需要一次遍历表计算得到,但我们通常不直接监控零值、缺失值率,因为不同特征上比率都不一样,A特征可能5%,B特征可能10%,直接对比率配置告警导致每个特征的阈值都不一样,配置复杂,因而我们监控零值缺失值率的波动(即当前周期零值缺失值比率同其他周期的差值),此时它们是衍生指标,因为波动的计算只依赖于当前和对比周期的零值、缺失值比率,同时对比周期的比率在历史任务上就已经完成计算,复用结果可以节省一倍以上的计算资源,提升效率。

除了核心模块,统一监控计算与检查工具还提供了“发出告警指令”、“接受告警反馈重新生成计算任务”等辅助模块。以上共同组成了统一监控计算与检查工具,确保触发的异常告警能够得到及时反馈修正。

部署视图

在实际部署上,统一监控计算与检查工具中TaskMaker(任务生成)、Calculator(计算)、Checker(检查)等模块实际上对应一个Spark节点,各个模块之间依赖关系如下图所示。

计算任务主要由TaskMaker模块根据用户配置生成,此外用户反馈已经修复的告警也会重新生成计算任务,TaskMaker屏蔽了不同调度周期的数据任务生成周期不一致的问题,例如日表任务每天生成前一天的表监控任务,月表任务只在每月特定一天生成月表的监控任务;Calculator接受计算任务完成监控指标的计算,Calculator完成监控的多数计算,需要较多的计算资源;Checker完成监控指标的衍生和检查。

模块详细设计

接下来,我们讨论监控工具TaskMaker(任务生成)、Calculator(计算)、Checker(检查)等模块的设计难点。

计算任务生成(TaskMaker)模块

计算任务生成(TaskMaker)模块核心逻辑是:

解析配置表 (配置表字段见下表);

根据配置表中schedule_type调度周期和schedule_bias偏置,生成需要检查表的分区(partition_name),将源表信息(table_name、col_name、partition_name)和计算信息(cal_procedure、kwargs)写回。

TaskMaker的主要设计难点在于:需要处理监控任务调度周期与源表计算任务调度周期的差别,适配好hour/day/week/month等不同周期,可以总结为下表:

假设当前数据时间是20210210 11:00,因为调度系统对日任务通常有一天的偏移,此时实际执行时间为20210211 11:00,对于不同类型的源表周期,偏置和最终生成分区举例如下:

hour: 如果偏置是-1,则检查分区和当前数据时间一致,为20210210 11:00,如果是-2,则检查分区提前一小时,为20210210 10:00;

day:如果偏置是-1,则检查分区和当前数据时间一致,为20210210,如果是-2,则检查分区提前一天,为20210209;

week: 如果偏置是-1,代表检查上一周,但是因为当天是周三,不生成周计算任务;

month:如果偏置是-10,生成上月计算任务202101,如果不是-10,则不生成月计算任务,注意到区别于小时表、日表、周表,偏置通常表示偏移若干个周期,但是月表例外,月表的偏置代表“几号开始计算任务”。

源表日表、月表等不同调度周期的问题在TaskMaker模块解决,后续模块不再感知源表周期的区别,专注完成监控指标的计算。

监控指标计算(Calculator)模块

监控指标计算(Calculator)模块核心逻辑(如下图)是:

读取未完成的计算任务;

通过生成执行计划并优化的方式,合并不同业务同学对同一表的监控计算需求,提升计算效率,Calculator会产生三个字段,分别为:

cal_time:保存计算时间

cal_outputs:保存计算结果,json格式

cal_errors:保存计算异常错误信息

首先,我们通过实例来解释如何通过执行优化避免重复计算,提升性能:

同学1的业务需要检查table表的A列的psi

同学2的业务需要检查table表的B列的psi

同学3的业务需要检查table表的C列缺失率占比

计算psi需要3次遍历表,计算缺失率需要1次遍历表,共计需要3+3+1=7次遍历

而实际上都是对同一table表的遍历,可以合并,如下图,此时只需要3次遍历,可以节省一倍以上的时间

为了实现执行优化,我们需要将一个监控指标的计算过程拆解为若干个最小可执行单元,称之为函数。具体到实现上,函数保存了计算逻辑的实现代码,过程调用若干个函数完成监控指标的最终计算,如下:

# Function
def F:RDD_aggre(...): ...
def F:math_psi(...): ...

# Procedure
def P:psi(...):
seg = F:RDD_aggre('cal_seg', ...)
cur = F:RDD_aggre('count@cur', seg, ...)
his = F:RDD_aggre('count@-1', seg, ...)
psi = F:math_psi(cur, seg, ...)
return psi

上述计算过程可以转换成计算图DAG来表示,如下图:

更复杂的,当有多个监控计算过程时,DAG可以表示为:

DAG需要执行的部分为叶子节点,为了避免重复计算,

我们对每次执行的叶子节点进行两类类优化:

合并同名函数,当函数名和参数都完全一致时,合并函数,仅执行一次;当函数名一致、参数不一致,生成新的执行函数(主要针对RDD aggregate操作),同样进执行一次。

缓存计算结果,缓存函数结果,当需要再次计算相同函数时,直接从缓存读取结果。

例如,上述DAG叶子节点表示的可执行函数(叶子节点)为:F:RDD_aggre(cal_seg,表1,A列)、F:RDD_aggre(cal_seg,表1,A列)、F:RDD_aggre(null_rate,表1,B列),其中两个F:RDD_aggre(cal_seg,表1,A列)为同名同参函数,合并为一个执行,又F:RDD_aggre(cal_seg,表1,A列)与F:RDD_aggre(null_rate,表1,B列)是同名函数,可以合并执行F:RDD_aggre([cal_seg, null_rate],[表1, 表1],[A列, B列]),此时原本需要需要三次遍历表,合并为一次遍历表即可完成。

同样的,在第二层叶子节点函数F:RDD_aggre(count@cur,seg, 表1,A列)可以合并为一次执行,但F:RDD_aggre(count@-1, seg,表1,A列)、F:RDD_aggre(count@-6, seg,表1,A列)需要分别遍历不同的表分区(上一周期分区、前6周期分区),因而只能分别计算,第二层叶子节点共产生三次遍历表,如下:

最终,拉取分段计数,在本地完成PSI的计算:

综上,执行优化算法小结如下:

算法: 执行优化算法。

输入:当前全部未执行计算任务对应计算过程。

流程:

- Step1. 将计算过程转化成DAG表示,每个节点为一个执行函数。

- Step2. 如果当前还存在未执行的叶子节点,那么合并叶子节点中的同名函数,当函数名和参数都完全一致时,合并函数;当函数名一致、参数不一致,生成新的执行函数。

- Step3. 执行函数,如果缓存中存在结果,直接拉取结果,否则完成计算后缓存结果。

- Step4. 若还存在未执行的叶子节点,返回Step2,否则终止。

输出:计算过程对应的监控指标结果。

当前,Calcutor支持常见监控指标包括:

通过Calcutor模块可以完成监控指标的计算,但也存在一些监控指标(如空值占比)需要衍生后才能判断是否异常,因而我们设计了Checker模块。

监控指标衍生与检查(Checker)模块

监控指标衍生与检查(Checker)模块核心逻辑为:

读取未检查的监控指标;

按gen_procedures衍生逻辑中配置方法对监控指标衍生后,按check_strategies检查逻辑中配置方法对监控指标检查;

Checker会产生五个字段,分别为:

check_time :保存计算时间

gen_outputs :保存衍生,json格式

gen_errors :保存衍生异常错误信息

check_pass:检查是否通过

check_details:保存未通过原因,接入方同学需关注check_pass、check_details字段

Checker的衍生方法包括:

完成衍生后,需要判断指标是否异常,这是通过检查阈值实现的,常见的检查逻辑有:

举个例子,如果要对缺失率波动进行监控,要求其变化幅度小于0.1,并且变化率小于0.2,可以将指标衍生配置为 diff@-1、relative@-1,指标检查策略配置为 abs_less_than@0.1 、 abs_less_than@0.2 。

监控计算优化实例 - PSI计算从20h到2h

在我们的实践中,发现对6w个数据列的psi等4个监控指标的计算,仅日表监控计算耗时长达20h+ ,计算耗时过大,长时间占用集群资源也会导致线上任务延迟。我们分析了造成计算时间长的原因有:

部分监控指标如PSI计算涉及多次遍历表;

Pyspark 原生Row属性访问效率差;

部分超大表行数达到20亿+。

针对这些问题,我们提出了下述方案逐一解决。

PSI计算优化:从4次遍历表到一次遍历表

相比缺失值占比、零值占比只需一次遍历表,计算psi@-1、psi@-6总共需要4次遍历表,具体如下:

遍历当前周期获取分段segs;

根据分段segs遍历当前周期获取分段计数;

根据分段segs遍历-1周期获取分段计数,计算psi@-1;

根据分段segs遍历-6周期获取分段计数,计算psi@-6。

为了降低PSI的遍历次数,我们设计了一种基于直方图的PSI估算方法,通过一次遍历表,得到特征分布直方图,再结合历史上计算的其他周期特征分布直方图,就可以估算出PSI。

如下图所示,基于直方图的PSI估算方法主要包括4个步骤:

- 步骤一:遍历一次表,使用蓄水池采样数据(>10w),本地计算分段、统计各个分段计数,得到特征的直方图分布h1,如下图;

- 步骤二:从历史结果中拉取-n周期的直方图分布h2;

- 步骤三:由于“分割点”不一致,我们无法直接根据直方图计算PSI,因此对直方图进行分割,使得当前周期直方图和上一周期直方图的分割点一致,取h1、h2直方图分割点的并集作为新分割点,按照新的分割点重新划分直方图得到h1`、h2`;

- 步骤四:根据分隔后的直方图h1`、h2`和PSI计算公式计算PSI即可。

通过PSI计算优化,计算时间从20h -> 7h。

Pyspark Row属性访问优化

我们发现Pyspark实现的Row访问属性有效率问题(如下图,官方源码注释也承认了这一问题),row['field']需要遍历所有的列名,才能得到正确的下标,其时间复杂度是O(n)。

为了解决row访问速度的问题,我们提出了下述方案:

广播[列名->列下标]Map:field_map = broadcast(field_map)

所有用row['field']的地方, 都改成 row[feld_map.value['field']]

通过使用了少量的内存存储[列名->列下标]映射,即能将Row属性访问复杂度从O(n) -> O(1),最终实验证明计算时间从7h -> 4h。

超大表的优化:采样与避免序列化

我们观察到,目前存在少量监控表行数达到20亿+,历史原因其格式为format(慢于orcfile),这些表全表遍历计算监控指标的时间达到数个小时。

针对这种超大表,我们提出了采样和避免序列化的优化方法,具体来说:

采样,即对行数大于1亿的表采样,控制行数在一亿内,需要注意的是,为了保证采样效率,我们使用where子句完成采样:where rand(123) < 一亿/表行数;

避免序列化,即通过DataFrame API where 或 select子句筛选不使用的行或列,避免它们序列化到Python对象。

通过上述优化,对于20亿+行数的大表计算时间从数个小时到几十分钟,并最终实现总体计算时间从20h -> 2h的优化。

小结

针对金融风控要素监控的“开发门槛高”“重复工作多”等问题,本文提出了“统一监控计算与检查工具”这一解决方案,本文详细论述了该方案TaskMaker、 Calculator、 Checker等各个模块的设计实现,最终实现了用户无需额外开发,只需要简单的配置即可完成监控指标的计算和推送,避免了人力浪费,提升效能。最后,我们还给出了一个“监控计算模块”优化的实例,通过“直方图估算PSI”、“Row列名广播”、“采样与避免序列化”等方式,将监控计算的速率提升了10倍,节省了大量计算资源。

 
   
414 次浏览       51
相关文章

微服务测试之单元测试
一篇图文带你了解白盒测试用例设计方法
全面的质量保障体系之回归测试策略
人工智能自动化测试探索
相关文档

自动化接口测试实践之路
jenkins持续集成测试
性能测试诊断分析与优化
性能测试实例
相关课程

持续集成测试最佳实践
自动化测试体系建设与最佳实践
测试架构的构建与应用实践
DevOps时代的测试技术与最佳实践
最新课程计划
 
最新文章
大数据平台测试
微服务架构下的测试之道
从零开始掌握微服务软件测试
如何进行测试需求分析:从接收需求到用例设计
python_selenium自动化测试框架
最新课程
测试需求分析与测试用例设计
性能测试方法与技术
自动化测试框架设计高级实践
接口自动化测试方法与工具
软件测试方法与实践(贯穿案例)
更多...   
成功案例
某支付企业 单元测试与重构培训
北京 用户体验、可用性测试与评估
某军工研究单位 自动化测试方法、案例与工具
知名消费金融公司 探索性测试与测试分析
北京 航天科工某子公司 软件测试架构师
更多...