【Flink银行反欺诈系统设计方案】1.短时间内多次大额交易场景的flink与cep的实现
【flink应用系列】1.Flink银行反欺诈系统设计方案
- 1. 经典案例:短时间内多次大额交易
- 1.1 场景描述
- 1.2 风险判定逻辑
- 2. 使用Flink实现
- 2.1 实现思路
- 2.2 代码实现
- 2.3 使用Flink流处理
- 3. 使用Flink CEP实现
- 3.1 实现思路
- 3.2 代码实现
- 4. 总结
1. 经典案例:短时间内多次大额交易
1.1 场景描述
规则1:单笔交易金额超过10,000元。
规则2:同一用户在10分钟内进行了3次或更多次交易。
风险行为:同时满足规则1和规则2的交易行为。
1.2 风险判定逻辑
检测每笔交易是否满足“单笔交易金额超过10,000元”。
对同一用户,统计10分钟内的交易次数。
如果交易次数达到3次或更多,则判定为风险行为。
2. 使用Flink实现
2.1 实现思路
使用Flink的KeyedStream按用户分组。
使用ProcessFunction实现自定义窗口逻辑,统计10分钟内的交易次数。
结合规则1和规则2,判断是否为风险行为。
2.2 代码实现
// 定义交易数据POJO
public class Transaction {private String transactionId;private String userId;private Double amount;private Long timestamp;// getters and setters
}// 定义风控结果POJO
public class RiskResult {private String userId;private String transactionId;private String riskLevel;private String actionTaken;private Long createTime;// getters and setters
}// 实现风控逻辑
public class FraudDetectionProcessFunction extends KeyedProcessFunction<String, Transaction, RiskResult> {private transient ValueState<Integer> transactionCountState;private transient ValueState<Long> timerState;@Overridepublic void open(Configuration parameters) {// 初始化状态ValueStateDescriptor<Integer> countDescriptor = new ValueStateDescriptor<>("transactionCount", Types.INT);transactionCountState = getRuntimeContext().getState(countDescriptor);ValueStateDescriptor<Long> timerDescriptor = new ValueStateDescriptor<>("timerState", Types.LONG);timerState = getRuntimeContext().getState(timerDescriptor);}@Overridepublic void processElement(Transaction transaction,Context ctx,Collector<RiskResult> out) throws Exception {// 规则1:单笔交易金额超过10,000元if (transaction.getAmount() > 10000) {// 更新交易次数Integer count = transactionCountState.value();if (count == null) {count = 0;}count += 1;transactionCountState.update(count);// 如果是第一次满足规则1,设置10分钟的定时器if (count == 1) {long timer = ctx.timestamp() + 10 * 60 * 1000; // 10分钟ctx.timerService().registerEventTimeTimer(timer);timerState.update(timer);}// 规则2:10分钟内交易次数达到3次if (count >= 3) {RiskResult result = new RiskResult();result.setUserId(transaction.getUserId());result.setTransactionId(transaction.getTransactionId());result.setRiskLevel("HIGH");result.setActionTaken("ALERT");result.setCreateTime(System.currentTimeMillis());out.collect(result);}}}@Overridepublic void onTimer(long timestamp, OnTimerContext ctx, Collector<RiskResult> out) throws Exception {// 定时器触发时,重置状态transactionCountState.clear();timerState.clear();}
}
2.3 使用Flink流处理
java
DataStream<Transaction> transactionStream = env.addSource(transactionSource);DataStream<RiskResult> riskResultStream = transactionStream.keyBy(Transaction::getUserId).process(new FraudDetectionProcessFunction());riskResultStream.addSink(new AlertSink());
3. 使用Flink CEP实现
Flink CEP(Complex Event Processing)是Flink提供的复杂事件处理库,适合处理基于时间序列的模式匹配。以下是使用Flink CEP实现上述风控规则的示例。
3.1 实现思路
定义模式:检测10分钟内3次或更多次大额交易。
使用Flink CEP的模式匹配功能,匹配符合条件的事件序列。
3.2 代码实现
java
// 定义交易数据POJO
public class Transaction {private String transactionId;private String userId;private Double amount;private Long timestamp;// getters and setters
}// 定义风控结果POJO
public class RiskResult {private String userId;private List<String> transactionIds;private String riskLevel;private String actionTaken;private Long createTime;// getters and setters
}// 实现风控逻辑
public class FraudDetectionCEP {public static void main(String[] args) throws Exception {StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();// 交易数据流DataStream<Transaction> transactionStream = env.addSource(transactionSource).assignTimestampsAndWatermarks(WatermarkStrategy.<Transaction>forBoundedOutOfOrderness(Duration.ofSeconds(5)).withTimestampAssigner((event, timestamp) -> event.getTimestamp()));// 按用户分组KeyedStream<Transaction, String> keyedStream = transactionStream.keyBy(Transaction::getUserId);// 定义CEP模式:10分钟内3次或更多次大额交易Pattern<Transaction, ?> pattern = Pattern.<Transaction>begin("first").where(new SimpleCondition<Transaction>() {@Overridepublic boolean filter(Transaction transaction) {return transaction.getAmount() > 10000;}}).next("second").where(new SimpleCondition<Transaction>() {@Overridepublic boolean filter(Transaction transaction) {return transaction.getAmount() > 10000;}}).next("third").where(new SimpleCondition<Transaction>() {@Overridepublic boolean filter(Transaction transaction) {return transaction.getAmount() > 10000;}}).within(Time.minutes(10));// 应用模式PatternStream<Transaction> patternStream = CEP.pattern(keyedStream, pattern);// 生成风控结果DataStream<RiskResult> riskResultStream = patternStream.process(new PatternProcessFunction<Transaction, RiskResult>() {@Overridepublic void processMatch(Map<String, List<Transaction>> match,Context ctx,Collector<RiskResult> out) throws Exception {RiskResult result = new RiskResult();result.setUserId(match.get("first").get(0).getUserId());result.setTransactionIds(match.values().stream().flatMap(List::stream).map(Transaction::getTransactionId).collect(Collectors.toList()));result.setRiskLevel("HIGH");result.setActionTaken("ALERT");result.setCreateTime(System.currentTimeMillis());out.collect(result);}});// 输出结果riskResultStream.addSink(new AlertSink());env.execute("Fraud Detection with Flink CEP");}
}
4. 总结
Flink实现:通过KeyedProcessFunction和状态管理实现多规则匹配。
Flink CEP实现:通过定义复杂事件模式,简化多规则匹配的逻辑。
适用场景:
Flink适合需要自定义逻辑的场景。
Flink CEP适合基于时间序列的模式匹配场景。
通过以上实现,可以高效检测银行交易中的风险行为,并根据需要扩展更多规则
相关文章:
【Flink银行反欺诈系统设计方案】1.短时间内多次大额交易场景的flink与cep的实现
【flink应用系列】1.Flink银行反欺诈系统设计方案 1. 经典案例:短时间内多次大额交易1.1 场景描述1.2 风险判定逻辑 2. 使用Flink实现2.1 实现思路2.2 代码实现2.3 使用Flink流处理 3. 使用Flink CEP实现3.1 实现思路3.2 代码实现 4. 总结 1. 经典案例:短…...
HashMap的table数组何时初始化?默认容量和扩容阈值是多少?
HashMap 的 table 数组何时初始化? 答案: table 数组在第一次调用 put() 方法时初始化。 为什么? HashMap 为了节省内存,采用了“懒加载”机制。即使用 new HashMap() 创建对象时,只是计算了参数(如容量、…...
基于CURL命令封装的JAVA通用HTTP工具
文章目录 一、简要概述二、封装过程1. 引入依赖2. 定义脚本执行类 三、单元测试四、其他资源 一、简要概述 在Linux中curl是一个利用URL规则在命令行下工作的文件传输工具,可以说是一款很强大的http命令行工具。它支持文件的上传和下载,是综合传输工具&…...
docker学习笔记(1)从安装docker到使用Portainer部署容器
docker学习笔记第一课 先交代背景 docker宿主机系统:阿里云ubuntu22.04 开发机系统:win11 docker镜像仓库:阿里云,此阿里云与宿主机系统没有关系,是阿里云提供的一个免费的docker仓库 代码托管平台:github&…...
数据集/API 笔记:新加坡PSI(空气污染指数)API
data.gov.sg 数据范围:2016年2月 - 2025年3月 1 获取API方式 curl --request GET \--url https://api-open.data.gov.sg/v2/real-time/api/psi 2 返回数据 API 的数据结构可以分为 3 大部分: 区域元数据(regionMetadata) →…...
计算机网络数据传输探秘:包裹如何在数字世界旅行?
计算机网络数据传输探秘:包裹如何在数字世界旅行? 一、从快递网络看数据传输本质 想象你网购了一件商品: 打包:商家用纸箱包装,贴上地址标签(数据封装)运输:包裹经过网点→分拣中心→运输车(网络节点与链路)签收:快递员核对信息后交付(数据校验与接收)数据的网络…...
笔记:代码随想录算法训练营day36:LeetCode1049. 最后一块石头的重量 II、494. 目标和、474.一和零
学习资料:代码随想录 1049.最后一块石头的重量II 力扣题目链接 思路:如何讲该问题转化为背包问题:还是对半分去碰,对半分去碰碰剩下的就是最小的。然后背包容量就是一半儿,物品重量等于物品价值等于stones[i] 和上…...
Bitmap -> Bitmap安卓设备上的显示和内存
Android 屏幕显示与 Bitmap 内存详解 前言 在 Android 开发中,理解屏幕显示单位和 Bitmap 内存占用是构建高效应用的基础。本文将详细介绍相关概念、计算公式及单位转换,并通过实例分析 Bitmap 在内存中的表现。 一、屏幕显示单位基础 1.1 基本单位及…...
QT study DAY2
作业 代码 Widget.h class Widget : public QWidget {Q_OBJECTpublic:Widget(QWidget *parent nullptr);~Widget();void save_data(const QString& filename,const QString& data); private slots:void on_lineEdit_textChanged(); //账户栏void on_l…...
QT-自定义参数设计框架软件
QT-自定义参数设计框架软件 Chapter1 QT-自定义参数设计框架软件前言一、演示效果二、使用步骤1.应用进行参数注册2.数据库操作单例对象3.参数操作单例对象 三、下载链接 Chapter2 Qt中管理配置参数(QSettings、单例模式)1 前言2 QSettings类ini文件写in…...
VUE集成Live2d
VUE集成Live2d 目前基于大模型,可以实现一个桌面的3D动画小人,个人猜测可以简介这个项目进行实现 1-参考网址 试了很多项目,只有这个项目直观的把问题说清楚了 Live2D Vue3技术应用:https://blog.csdn.net/hh1233321/article/details/1406947…...
【CPP面经】科大讯飞 腾讯后端开发面经分享
文章目录 C 面试问题整理基础问题简答1. 内存对齐2. this 指针3. 在成员函数中删除 this4. 引用占用内存吗?5. C 越界访问场景6. 进程通信方式7. 无锁队列实现8. ping 在哪一层?实现原理?9. HTTPS 流程10. GDB 使用及 CPU 高使用定位11. 智能…...
el-card 结合 el-descriptions 作为信息展示
记录下el-card 组合 el-descriptions 实现动态展示信息 文章结构 实现效果1. el-descriptions 组件使用1.1 结合v-for实现列表渲染1.2 解析 2. 自定义 el-descriptions 样式2.1 修改背景色、字体颜色2.2 调整字体大小2.3 解析 3. el-card 结合 el-descriptions 作为信息展示3.…...
GaussDB自带诊断工具实战指南
一、引言 GaussDB是一种分布式的关系型数据库。在数据库运维中,快速定位性能瓶颈、诊断故障是保障业务连续性的关键。GaussDB内置了多种诊断工具,结合日志分析、执行计划解析和实时监控功能,帮助开发者与运维人员高效解决问题。本文深入讲解…...
LeetCode 链表章节
简单 21. 合并两个有序链表 将两个升序链表合并为一个新的 升序 链表并返回。新链表是通过拼接给定的两个链表的所有节点组成的。 示例 1: 输入:l1 [1,2,4], l2 [1,3,4] 输出:[1,1,2,3,4,4]示例 2: 输入:l1 [], l2…...
SSL证书和HTTPS:全面解析它们的功能与重要性
每当我们在互联网上输入个人信息、进行在线交易时,背后是否有一个安全的保障?这时,SSL证书和HTTPS便扮演了至关重要的角色。本文将全面分析SSL证书和HTTPS的含义、功能、重要性以及它们在网络安全中的作用。 一、SSL证书的定义与基本概念 S…...
正交投影与内积空间:机器学习的几何基础
前言 本文隶属于专栏《机器学习数学通关指南》,该专栏为笔者原创,引用请注明来源,不足和错误之处请在评论区帮忙指出,谢谢! 本专栏目录结构和参考文献请见《机器学习数学通关指南》 正文 🔍 1. 内积空间的…...
Qt中txt文件输出为PDF格式
main.cpp PdfReportGenerator pdfReportGenerator;// 加载中文字体if (QFontDatabase::addApplicationFont(":/new/prefix1/simsun.ttf") -1) {QMessageBox::warning(nullptr, "警告", "无法加载中文字体");}// 解析日志文件QVector<LogEntr…...
《HelloGitHub》第 107 期
兴趣是最好的老师,HelloGitHub 让你对编程感兴趣! 简介 HelloGitHub 分享 GitHub 上有趣、入门级的开源项目。 github.com/521xueweihan/HelloGitHub 这里有实战项目、入门教程、黑科技、开源书籍、大厂开源项目等,涵盖多种编程语言 Python、…...
Langchain解锁LLM大语言模型的结构化输出能力(多种实现方案)
在 LangChain解锁LLM大语言模型的结构化输出能力:调用 with_structured_output() 方法 这篇博客中,我们了解了格式化LLM输出内容的必要性以及如何通过调用langchain框架中提供的 with_structured_output() 方法对LLM输出进行格式化(三种可选方…...
Java 语言特性(面试系列2)
一、SQL 基础 1. 复杂查询 (1)连接查询(JOIN) 内连接(INNER JOIN):返回两表匹配的记录。 SELECT e.name, d.dept_name FROM employees e INNER JOIN departments d ON e.dept_id d.dept_id; 左…...
树莓派超全系列教程文档--(62)使用rpicam-app通过网络流式传输视频
使用rpicam-app通过网络流式传输视频 使用 rpicam-app 通过网络流式传输视频UDPTCPRTSPlibavGStreamerRTPlibcamerasrc GStreamer 元素 文章来源: http://raspberry.dns8844.cn/documentation 原文网址 使用 rpicam-app 通过网络流式传输视频 本节介绍来自 rpica…...
智慧工地云平台源码,基于微服务架构+Java+Spring Cloud +UniApp +MySql
智慧工地管理云平台系统,智慧工地全套源码,java版智慧工地源码,支持PC端、大屏端、移动端。 智慧工地聚焦建筑行业的市场需求,提供“平台网络终端”的整体解决方案,提供劳务管理、视频管理、智能监测、绿色施工、安全管…...
【JVM】- 内存结构
引言 JVM:Java Virtual Machine 定义:Java虚拟机,Java二进制字节码的运行环境好处: 一次编写,到处运行自动内存管理,垃圾回收的功能数组下标越界检查(会抛异常,不会覆盖到其他代码…...
STM32标准库-DMA直接存储器存取
文章目录 一、DMA1.1简介1.2存储器映像1.3DMA框图1.4DMA基本结构1.5DMA请求1.6数据宽度与对齐1.7数据转运DMA1.8ADC扫描模式DMA 二、数据转运DMA2.1接线图2.2代码2.3相关API 一、DMA 1.1简介 DMA(Direct Memory Access)直接存储器存取 DMA可以提供外设…...
Android 之 kotlin 语言学习笔记三(Kotlin-Java 互操作)
参考官方文档:https://developer.android.google.cn/kotlin/interop?hlzh-cn 一、Java(供 Kotlin 使用) 1、不得使用硬关键字 不要使用 Kotlin 的任何硬关键字作为方法的名称 或字段。允许使用 Kotlin 的软关键字、修饰符关键字和特殊标识…...
学校时钟系统,标准考场时钟系统,AI亮相2025高考,赛思时钟系统为教育公平筑起“精准防线”
2025年#高考 将在近日拉开帷幕,#AI 监考一度冲上热搜。当AI深度融入高考,#时间同步 不再是辅助功能,而是决定AI监考系统成败的“生命线”。 AI亮相2025高考,40种异常行为0.5秒精准识别 2025年高考即将拉开帷幕,江西、…...
基于IDIG-GAN的小样本电机轴承故障诊断
目录 🔍 核心问题 一、IDIG-GAN模型原理 1. 整体架构 2. 核心创新点 (1) 梯度归一化(Gradient Normalization) (2) 判别器梯度间隙正则化(Discriminator Gradient Gap Regularization) (3) 自注意力机制(Self-Attention) 3. 完整损失函数 二…...
CSS | transition 和 transform的用处和区别
省流总结: transform用于变换/变形,transition是动画控制器 transform 用来对元素进行变形,常见的操作如下,它是立即生效的样式变形属性。 旋转 rotate(角度deg)、平移 translateX(像素px)、缩放 scale(倍数)、倾斜 skewX(角度…...
MinIO Docker 部署:仅开放一个端口
MinIO Docker 部署:仅开放一个端口 在实际的服务器部署中,出于安全和管理的考虑,我们可能只能开放一个端口。MinIO 是一个高性能的对象存储服务,支持 Docker 部署,但默认情况下它需要两个端口:一个是 API 端口(用于存储和访问数据),另一个是控制台端口(用于管理界面…...
