对信用卡欺诈 Say No|百行代码实现简化版实时欺诈检测

Posted woqutechteam

tags:

篇首语:本文由小常识网(cha138.com)小编为大家整理,主要介绍了对信用卡欺诈 Say No|百行代码实现简化版实时欺诈检测相关的知识,希望对你有一定的参考价值。

进入互联网时代,你的绝大部分操作都可以在网上进行,极大的方便了我们的生活。但是信用卡盗刷者也可以利用网络来诈骗,典型的做法是:诈骗者首先入侵安全级别较低系统来盗窃信用卡卡号,用盗得的信用卡进行很小额度的消费进行测试,如果测试消费成功,那么他们就会用这个信用卡进行大笔消费,来购买那些可以倒卖的财物,实现诈骗敛财的目标。

大部分银行都有针对信用卡诈骗的反欺诈检测系统,通过对诈骗模式进行识别,及时通知用户或者直接冻结账户,来避免进一步损失。flink的入门介绍文档中就展示一种信用卡诈骗检测的实现方式,但是数据来源是一个静态数组,不符合实际用户场景。本文将介绍一种方案,通过少于 100 行代码修改该实例程序,实现基于 Oracle 上账户表变更的实时欺诈检测。

1.传统实时欺诈检测方案分析

Flink示例程序会检测每一笔交易,若发现一个帐户在 1 分钟内,先出现了一笔小交易(小于 1),后面又出现了一笔大交易(大于 500),则认为出现了欺诈交易,立即输出警告。具体的代码解析可以阅读基于DataStream API 实现欺诈检测

但是,示例程序中的数据来源 TransactionSource 的数据来源是一个静态数组 private static List<Transaction> data = Arrays.asList(new Transaction(1L, 0L, 188.23D)...的迭代器。一般情况下,客户的余额是存储在 Oracle 的账户表中的。怎么将客户余额的变化输出到 Flink 中,来实现实时的欺诈检测列?能想到的方案列举如下:

  • 方案 1:轮询从 Oracle 账户表查询余额变更 应用程序固定时间间隔去轮询 Oracle 账户表的数据,检查到某个客户的账户余额发生了变化后,通知 Flink 进行欺诈检测。这种方案需要不断轮询 Oracle 数据库,对有数据库性能影响,并且就算轮询的间隔足够短,还是有可能漏掉了一些账户变更信息,不可取。

  • 方案 2:业务代码修改 Oracle 账户表时通知 Flink 修改交易程序,在它去更新 Oracle 账户表时,通知 Flink 进行欺诈检测。这种方案的优势在于不会丢掉任何账户变更的事件;但是需要修改交易程序,会导致业务程序耦合度提升。实现上如果采用同步模式,可能会由于 Flink 失败而导致交易失败,也会大大提高交易持续时间;而采用异步方式,需要考虑通知 Flink 和写入账户表的原子性,有可能成功通知了 Flink 但是写入账户表失败了,也有可能写入账户表成功了,却没有通知到 Flink。

  • 方案 3:利用 logminer 抽取账户表变更 Oracle 提供 logminer 来将数据库日志反解析成变更 SQL,这样就可以将 Oracle 账户表更新的信息抽取出来,通知 Flink 进行欺诈检测。这种方案的优点在于直接基于 Oracle 数据表的修改来做增量的同步(oracle 日志中记录账户表修改并提交了,说明客户修改账户是成功的,不用担心 Flink 通知了,账户表反而写失败了),降低了业务的耦合度,也不会担心丢失了账户变更事件;但是 logminer 每次只能挖掘一整个日志的变化,没法断点续传,并且挖掘的数据也只能写入 alert.log,会污染错误日志。这个方案缺陷也比较大。

上述三个方案都有一定的缺陷和问题,要么可能会漏掉部分变更数据,要么可能影响 oracle 性能。logminer 相对来说是避免漏数据,对数据库性能影响最小的方案,是否有一个类似于 logminer 而且支持断点续传,对 Flink 又比较友好的方案?

  • 方案 4:利用 OGG 抽取账户表变更 Oracle 公司的 OGG 和国内部分厂商基本能避免 logminer 的缺点,但是需要在 Oracle 服务器上安装客户端,有侵入,并且配置和使用比较复杂,价格上也不是很友好。

最后我们找到了一个轻量、免费日志解析工具QDecoder来替代 OGG,将 oracle 账户表的变更通知到 Flink,从而实现欺诈检测的方法。官方介绍:QDecoder 是沃趣科技自主研发,基于 Oracle redo 日志进行二进制解析的订阅同步工具,易集成、零侵入、高性能、全免费。目前,QDecoder 已经在多家证券和银行上线使用,稳定运行,得到诸多客户的肯定与认可。

2.Oracle 源端增量输出

期待实现的实时欺诈检测的架构:

如图:

  • QDecoder 实时获取 Oracle 的 redo log,将账户表数据变更解析出来,写入 kafka 的指定 topic 中。

  • Flink 程序(FraudDetection)使用 flink-connector-kafka 从 kafka 获取交易数据,进行流式计算,识别出可能的欺诈交易,并输出警告。

2.1 安装 QDecoder

为了从 Oracle 的日志中挖掘出 account 表的变更数据,我们需要安装 QDecoder。QDecoder 的安装也非常简单,并且它跟其他的数据库同步软件不一样,它默认是不需要到 Oracle 服务器上去部署客户端的,只需要给定 ASM 账号就可以通过网络连接到 Oracle 上取日志。

一键安装命令如下: docker run -it --name=qdecoder -p 9191:9191 -p 9092:9092 --pull always registry.cn-hangzhou.aliyuncs.com/woqutech/qdecoder

根据提示配置 QDecoder,更多信息可参考 Docker Hub

以下配置需要特别注意:

  • 配置项 1.1 中列出的 sql,请以 dba 权限在 oracle 中执行,这将配置 QDecoder 查询系统表需要的权限。

  • 配置项 2.1: 输入将要检测的表:qdecoder.account。简单起见,我们这里把 account 示例表新建在 qdecoder 账户下了。conn qdecoder/qdecoder;create table account(accountid int primary key, balance number);

  • 配置项 3.1: 选择输出到 kafka, bootstrap.servers 可以不输入,直接在容器中启动 kafka,qdecoder 会将变更数据写入“defaultapp.qdecoder.binlog.qdecoder”的 topic。生产环境下如果有 kafka 集群,请输入集群连接的地址和账号。

下面是配置示例: 

2.2 更新账户表观察 QDecoder 的输出

等 QDecoder 成功启动后,可以按照提示运行 binlogdumpK,从 kafka 读取 binlog 并打印出来。由于 account 表并没有更新,此时没有变更数据输出。

我们来更新 account 表: 

insert into account values(1,10000);insert into account values(2,20000);insert into account values(3,30000);commit;

binlogdumpK 输出的 binlog 如下图:

上图中 schemaName: "qdecoder",tableName: "account", eventType: INSERT 表示这里是对qdecoder.account表的 INSERT 操作。其实它对应的 sql 就是insert into qdecoder.account(accountid,balance) values(1,10000)

这里

  • Oracle 的日志解析出来是遵循阿里巴巴canal的 protobuf 格式,这个是关系型数据库增量输出的标准格式

  • binlogdumpK 只是为了观察一下 QDecoder 的输出,你可以随时关掉它,这并不影响 QDecoder 和 Flink 程序的运行。

至此,我们已经利用 QDecoder 从 Oracle 的日志解析出账户表的数据变更,那么,怎么将这些输出作为 Flink 现有的欺诈检测的输入源列呢?

3.实现 Kafka Consumer 对接 Oracle 源端增量数据

 Flink示例程序是利用 addSource(new TransactionSource())来将静态数组作为源加入流处理的

DataStream<Transaction> transactions = env
            .addSource(new TransactionSource())
            .name("transactions");

我们要修改为从 kafka 中取日志,可以利用 DataStream Connectors 配置一个 Kafka Consumer。按照Flink自带的示例中的DataStream<String> stream = env.addSource(new FlinkKafkaConsumer<>("topic", new SimpleStringSchema(), properties));添加数据源。这里

  • "topic"应该修改为"defaultapp.qdecoder.binlog.qdecoder",这是 QDecoder 插入 kafka 的 topic 名称

  • properties 修改为你连接的 kafka 的地址,我们这里使用的是 QDecoder 自带的 kafka,修改为 127.0.0.1:9092

  • SimpleStringSchema()是直接以 string 的方式输出出来,QDecoder 的输出是 protobuf 的,不能直接用 string 的方式输出,而需要解析出来转换成欺诈检测认识的 Transaction

3.1 binlog 转换成 Transaction 实现

为了能完全重用 Flink 示例程序中的代码,我们这里需要

  • 使用 com.alibaba.otter.canal.protocol 反序列化 QDecoder 插入到 kafka 的 binlog 日志

  • 将 binlog 日志中的账户表变更转换成 Transaction 对象。

我们实现一个BinlogTransactionSchema类反序列化 binlog, 计算 balance 的变化,生成 org.apache.flink.walkthrough.common.entity.Transaction 对象。

主要代码如下:

// 反序列化Entry
CanalEntry.Entry entry = CanalEntry.Entry.parseFrom(binlog);
// 获取表名
entry.getHeader().getTableName();
// 获取Entry type:ROWDATA|TRANSACTIONBEGIN|TRANSACTIONEND|...
entry.getEntryType();
// 获取row change
CanalEntry.RowChange rowChange = CanalEntry.RowChange.parseFrom(entry.getStoreValue());
// 获取Event type: INSERT|UPDATE|DELTE|...
rowChange.getEventType();
// 获取执行时间
long executeTimeMs = entry.getHeader().getExecuteTime();
// 获取row data
CanalEntry.RowData rowData = rowChange.getRowDatas(0);

// 获取旧值
for (CanalEntry.Column col : rowData.getBeforeColumnsList()) 
    if (col.getName().equalsIgnoreCase("accountid"))  
       oldRow.accountId = Long.parseLong(col.getValue()); 
     else if (col.getName().equalsIgnoreCase("balance")) 
        oldRow.balance = Double.parseDouble(col.getValue());
    


// 获取新值
for (CanalEntry.Column col : rowData.getAfterColumnsList()) 
    if (col.getName().equalsIgnoreCase("accountid")) 
        newRow.accountId = Long.parseLong(col.getValue())
     else if (col.getName().equalsIgnoreCase("balance")) 
        newRow.balance = Double.parseDouble(col.getValue());
    


// 创建transactionnew
 Transaction(newRow.accountId, executeTimeMs, Math.abs(newRow.balance-oldRow.balance));

3.2 将数据源加入 Flink

QDecoder 输出的增量数据转换后,可以直接作为数据源形成 DataStream

修改后代码如下:

public static void main(String[] args)throws Exception
StreamExecutionEnvironment env=StramExecutionEnvironment.getExecutionEnvironment();

// 创建FlinkKafkaConsumer,用BinlogTransactionSchema反序列化
Properties properties = new Properties();
properties.setProperty("bootstrap.servers", "127.0.0.1:9092");
properties.setProperty("group.id", "flink.test");

// 指定"defaultapp.qdecoder.binlog.qdecoder"的topic,使用BinlogTransactionSchema来将QDecoder账户表的更新转换成Transaction
FlinkKafkaConsumer<Transaction> kafkaSource = new FlinkKafkaConsumer<Transaction>("defaultapp.qdecoder.binlog.qdecoder", new BinlogTransactionSchema(), properties);
kafkaSource.setStartFromEarliest();

DataStream<Transaction> transactions = env
    .addSource(kafkaSource)
    .name("transactions");
…

3.3 验证运行

修改好的应用程序我们已经上传到 github 上供下载,测试运行步骤如下:

3.3.1 程序下载

git clone https://github.com/woqutech/qdecoder.gitcd qdecoder/FlinkSample/frauddetection

3.3.2 程序运行

frauddetection 是一个 maven 创建的项目,有 pom.xml 项目文件,可以导入各种 IDE,进行调试和运行。这里只介绍大家常用的 intellij IDEA 运行验证方法

  • 打开项目 开始界面:open or import -> 选择 frauddetection 目录 或者 菜单: file/open -> 选择 frauddetection 目录

  • 运行程序 菜单: run -> run 'FraudDetectionJob'

注意:如果报告 slf4j 重复,且有大量的 log 输出,请在 module/dependencies 中删除 ch.qos.logback:logback-classic 和 ch.qos.logback:logback-core。

3.3.3 欺诈验证检测

QDecoder 和 frauddetection 正常运行以后,我们就可以更新 account.balance,模拟交易,来观察 frauddetection 程序是否能进行欺诈检测了。

  • 在 Oracle 上执行以下 SQL

update account set balance = balance - 0.1 where accountid = 1;commit;update account set balance = balance - 0.2 where accountid = 1;commit;update account set balance = balance + 100 where accountid = 2;commit;update account set balance = balance - 501 where accountid = 1;commit;update account set balance = balance - 200 where accountid = 2;commit;

这里我们模拟了两个账号 account=1/2 的交易。其中 account=2 先入账 100 元,然后扣了 200 元是正常交易,不满足欺诈检测条件。account=1 的账户余额先扣了 0.1 元,然后再扣了 0.2 元,最后直接扣了 501 元,满足欺诈检测的条件:在一分钟内(private static final long ONE_MINUTE = 60 * 1000;)先做小于 1 元(private static final double SMALL_AMOUNT = 1.00)的小额交易,然后再做 500 以上的大额交易(private static final double LARGE_AMOUNT = 500.00;)。

  • 检查欺诈交易是否被正确识别

执行完上述 SQL,frauddetection 程序会立即输出:

21:11:20,107 INFO  org.apache.flink.walkthrough.common.sink.AlertSink           [] - Alertid=1

表示 accountid=1 的帐号检测到欺诈交易。

至此,我们的 frauddetection 欺诈检测程序就修改完成了。

回顾整个过程:

  • 利用 QDecoder 实现了 Oracle 数据库的日志订阅增量数据导出

  • 利用 alibaba canal 的 protobuf 来解析 Oracle 增量变化

  • 实现了不到 100 行代码的 BinlogTransactionSchema 类,用于将解析数据转换为欺诈检测识别的数据

  • 利用 Flink 的 DataStream Connectors 从 kafka 取出增量变化数据

  • 利用了 Flink 实现增量流数据的有状态计算分布式处理,实现欺诈检测

4.总结

综上所述,我们利用 QDecoder 和 Flink 写了少于 100 行的代码,实现了一个简化版的银行信用卡欺诈检测程序。整体来说,利用 Flink+QDecoder 可以很容易将 oracle 的增量变化取出来,同步给大数据平台或者数据湖,有助于将静态数据流动起来,帮助企业盘活数据资产,提升运营决策效果。

以上是关于对信用卡欺诈 Say No|百行代码实现简化版实时欺诈检测的主要内容,如果未能解决你的问题,请参考以下文章

一百行代码实现简易版 ChatGPT | 社区征文

一百行代码实现简易版 ChatGPT 聊天机器人

一百行代码实现简易版 ChatGPT 聊天机器人

数据分析-信用卡反欺诈模型

信用卡欺诈的 Python 算法

阿里云堡垒机V3版重磅发布,和运维失误say no!