(19)国家知识产权局 (12)发明 专利申请 (10)申请公布号 (43)申请公布日 (21)申请 号 202211108450.3 (22)申请日 2022.09.09 (71)申请人 武汉众邦银行股份有限公司 地址 432200 湖北省武汉市黄陂区盘龙城 经济开发区汉口北大道88号汉口北国 际商品交易中心D2区1-2层、 2 2-23层 (72)发明人 汪大磊　徐浩　田羽　兰翔　 吴金林　 (74)专利代理 机构 成都正煜知识产权代理事务 所(普通合伙) 51312 专利代理师 徐金琼 (51)Int.Cl. G06F 9/448(2018.01) G06F 16/25(2019.01) G06F 16/27(2019.01)G06F 16/28(2019.01) (54)发明名称 基于FlinkSQL的风控指标计算开发的自动 配置处理方法及系统 (57)摘要 本发明公开了一种基于FlinkSQL的风控指 标计算开发的自动配置处理方法及系统， 属于 Flink实时批量数据处理技术领域， 解决现有技 术在Flink应用开发过程中多种数据源、 多种结 构类型数据和多种数据落地系统的配置和代码 冗余杂乱， 更新迭代频繁， 耦合度高造成的开发 效率降低和开发成本增加的问题。 本发明包括读 取数据流程、 数据处理和转换数据落地。 本发明 用于针对Flink应用开发过程中自定义源数据， 数据落地的配置及调用冗乱问题及处理不同结 构数据分发处 理。 权利要求书6页 说明书11页 附图2页 CN 115469941 A 2022.12.13 CN 115469941 A 1.一种基于FlinkSQL的风控指标计算开发的自动配置处理方法， 其特征在于， 包括如 下步骤： 步骤1.读取 数据流程： 根据数据源传入数据源类型参数， 从数据源类型参数的配置参数据或外部配置文件中 读取配置参数， 并基于flink自带的已经继承了数据源的消费接口获取数据源连接逻辑来 根据配置参数获得消费者实例对象来读取数据， 并返回Flink数据集， 其中， 数据源为Kafka 或关系型数据库， 关系型数据库包括mys ql和oracle， Flink数据集是Flink应用程序中的中 间数据结果， 包括DataStream数据集或DataSet数据集， 最初的数据集是读取数据源后产 生； 步骤2.数据处 理转换： 基于风控引擎系统获取风控JSON报文中的属性， 并将属性与逻辑字段名进行映射并存 放在配置表中， 基于配置表、 s ql脚本和给定的伪sql脚本解析得到JSON的java解析代码， 并 基于JSON的java解析代码和基于步骤1读取风控引擎系统中的数据源的topic引擎数据得 到包含topic引擎字段的数据的Flink数据集注册为Flink  table， 并对应到伪sql脚本中的 表名， 返回Fl ink数据集； 步骤3.数据落 地： 基于步骤2得到的Fl ink数据集传入配置参数据进行 数据落地。 2.根据权利要求1所述的一种基于FlinkSQL的风控指标计算开发的自动配置处理方 法， 其特征在于， 所述 步骤1的具体步骤为： 步骤1.1定义 开发用于数据源读取接口 的数据源分发类及方法： 定 义 数 据 源 分 发 类 来区 分 批处 理 及流 处 理 ， 数 据 源 分 发 类 包 括 批处 理 BatchSourceData和流处 理StreamSourceDatas； 数据源分发方法getData( )， 用于基于批处理BatchSourceData或流处理 StreamSourceDatas数据源类型参数 的传入， 调用数据源模板类方法init()， 并解析配置 文件， 其具体过程为： 根据批处理BatchSourceData和流处理StreamSourceDatas的传入的 数据源类型参数， 基于数据源类型参数调用flink连接数据源有内置的接口来调用相应数 据源模板类方法init()， flink连接数据源有内置的接口即指flink自带继承有数据源的 消费接口， 消费接口指数据源与Fl ink集成的连接器， 通过配置参数 得到消费者实例对象； 步骤1.2定义 开发数据源启动方法： 由于flink自带继承有数据源的消费接口， 则定义数据源启动方法onStart()用于获取 数据源连接逻辑， 无需调用数据源的消费接口， 即不需要重写读取数据的操作逻辑， 具体 为： 数据源在前期获取数据库driver的参数建立连接及读数准备逻辑中已定义SQL获取需 要读取的表和字段， 在 数据源启动方法onStart()中将定义好的数据读取逻辑传入执行方 法ResultSet中， 返回Fl ink数据集； 步骤1.3基于步骤1.1 ‑步骤1.2定义 开发用于数据源配置的数据源 模板类及方法： 定义数据源模板类来区分批处理及流处理两种， 数据源模板类包括批处理的数据流 BatchMysqlSourceData和流处 理的数据源St reamKafkaSourceData； 数据源模板类的方法init()， 用于接收数据源分发方法getdata()传入的批处理的数 据流BatchMysqlSourc eData或流处理的数据源StreamKafkaSourc eData的数据源类型参数权　利　要　求　书 1/6 页 2 CN 115469941 A 2来调用数据源模板类的方法init()， 数据源模板类方法init()负责调用数据源启动方法 onStart()来调用执行方法ResultSet解析数据源类型参数的配置参数据或外部配置文件 中， 以获取解析好的配置参数， 并调用flink连接数据源有内置的接口返回批处理或者流处 理的Flink数据集； 步骤1.4定义 开发具体数据源类及方法： 基于步骤1.1 ‑步骤1.3具体数据源类， 即基于Flink应用调用A数据源， 通过数据源分发 方法getData()传入的A数据源的类型的数据源类型参数， 基于A数据源的类型的数据源类 型参数调用数据源模板类方法init()， 数据源模板类方法init()负责调用数据源启动方 法onStart()来调用执行方法ResultSet解析A数据源的数据源类型参数的配置参数据或 外部配置文件中， 以获取解析好的配置参数将配置参数传入数据源模板类方法init()中， 接着传入flink中内置的用于读取A数据源的接口创建A数据源的消费者实例 对象， 然后调 用flink的核心方法addSource将消费者实例对象作为参数传入读取数据， 数据读取后返回 结果集， 其中， 结果是数据库查询的一个数据集合， A数据源为Kafka、 mysql或oracle数据 源。 3.根据权利要求2所述的一种基于FlinkSQL的风控指标计算开发的自动配置处理方 法， 其特征在于， 所述步骤1.2中的数据源启动方法onStart()还定义有数据源异常处理及 后置处理方法： 数据源异常处理方法： 在获取数据源及读取数据源过程中， 使用try  catch捕获异常， 包括连接超时、 配置异常或数据源异常； 后置处理方法： 数据源连接成功并且数据读取完成后 ， 调用数据库驱动创建预编译语句 PrepareStatement和连接的接口名Connection的close()方法关闭预编译语句和连接， 以 回收资源。 4.根据权利要求3所述的一种基于FlinkSQL的风控指标计算开发的自动配置处理方 法， 其特征在于， 所述 步骤2的具体步骤为： 步骤2.1定义源字段与逻辑字段间的映射： 通过风控引擎系统配置获取风控JSON报文中的属性， 根据业务逻辑定义逻辑字段名， 获取对应渠道及各级路径， 将JSON报文中的属性对应到逻辑字段名， 根据完整路径获取各 层级结构、 JSON对象名及 JSON对象数组名， 其中， JSON属性能对应到A层级的json对象， A层 级的json对象能对应到B层级的json对象， 依此类推， 根据完整路径即可获取多层级结构， 风控JSON报文中的属性包括数据源类型和shcema信息， shcema信息是源数据的元数据信 息； 步骤2.2将映射关系配置到数据库中， 即将步骤2.1中得到的各层级、 JSON对象名及 JSON对象数组名映射关系配置 到、 用于存放JSON数据结构的数据库中的配置表中； 步骤2.3开发自动化脚本， 封装解析逻辑： 开发计算转换及聚合逻辑形成sql脚本封装到用于存储sql脚本的函数中， 基于给定的 伪sql脚本(业务人员分析业务指标的给定的数据， 如学生、 老师等)采用封装后的sql脚本 在配置表中解析需要的逻辑字段得到JSON的java解析代码， 作为函数返回的结果输出， 其 中， 封装后的sql脚本和java解析代码存 储在Flink算子中；权　利　要　求　书 2/6 页 3 CN 115469941 A 3