Flink双流实时对账

Stella981
• 阅读 589

Flink双流实时对账

背景

在电商、金融、银行、支付等涉及到金钱相关的领域,为了安全起见,一般都有对账的需求。

比如,对于订单支付事件,用户通过某宝付款,虽然用户支付成功,但是用户支付完成后并不算成功,我们得确认平台账户上是否到账了。

针对上述的场景,我们可以采用批处理,或离线计算等技术手段,通过定时任务,每天结束后,扫描数据库中的数据,核对当天的支付数据和交易数据,进行对账。

想要达到实时对账的效果,比如有的用户支付成功但是并没有到账,要及时发出报警,我们必须得依赖实时计算框架。

我们将问题简单化,比如有如下场景,在某电商网站,用户创建订单并支付成功,会将相关信息发给kafka,字段包括,用户uid、动作、订单id、时间等信息

{userId=1, action='create', orId='order01', timestamp=1606549513} 

支付成功并且金额已经进入平台账户,往往也会把相关信息发给kafka,如订单id,支付方式、时间等信息。

{orId='order01', payChannel='wechat', timestamp=1606549536}

只有订单在支付(action=pay)成功后,并且成功到账,这才算一次完整的交易。本案例,就是要实时检测那些不成功的交易,如有不成功的,及时发出报警信息。

上述行为本身会产生两种事件流,一种是订单事件流,另一种是交易事件流,我们通过Flink将两种类型的流进行关联,实时分析没有到账的数据,发出报警。

为了简化,我们从socket读取数据流,代替从kafka消费数据。

代码示例

本案例涉及到的知识点:

  • 状态编程

  • 定时器

  • 延迟事件处理

  • 合流操作

首先,我们需要定义订单事件OrderEvents和交易事件ReceiptEvents

// 订单事件

通过Flink程序,联合两条流,实时检测交易失败的数据并输出到侧输出流里。

public static void main(String[] args) throws Exception {

上面代码的主要逻辑是:

  • 从端口为8888和9999的两个socket读取订单事件和交易事件(模拟从kafka消费),然后将事件数据包装成OrderEvents和ReceiptEvents。

  • 提取事件时间。

  • 对于OrderEvents,只需要action=pay的数据,过滤无用的数据。

  • 将两条流根据orId keyby,生成orderStream和receiptStream,并通过connect合并两条流,将合并后的结果,交给CoProcessFunction函数计算。

  • 将正常交易事件输出在success中,异常的交易事件,输出到两个侧输出流中。

所以,我们需要自定义聚合函数,继承CoProcessFunction函数,实现正常交易和异常交易行为的实时计算。

class MyCoProcessFunction 

上述代码是我们自定义的窗口函数,主要的功能是:

  • 继承了CoProcessFunction,分别在processElement1和procossElement2方法中处理orderEvents和receiptEvent。

  • 定义状态,侧输出流,注册定时器,通过一些逻辑计算是是正常交易还是异常交易。

  • 在processElement1方法中,如果只有pay事件没有receipt事件,则注册一个5s后触发的定时器,等待receipt事件的到来,如果5s后receipt事件仍没有到来,则说明是一个异常交易事件,触发timer并将异常事件输出到侧输出流中。

  • 在processElement2方法中,如果只有receipt事件没有pay事件,表明pay事件和receipt事件乱序,则注册一个3s的定时器,等待pay事件。如果3s后还是没有pay事件到达,则触发timer将延迟的乱序数据输出到侧输出流中。

  • 定义定时器timer,对于异常的交易行为,将交易输出输出到侧输出流。异常交易是指,在一定时间范围内,只有pay事件没有receipt事件 或 只有receipt事件没有pay事件。如果在一定时间范围内这两个事件都有,则属于正常交易行为。

打开两个socket,输入数据模拟交易行为。为了输出一些异常信息,我们的输入方式,不光要正常输入数据,还要输入一些乱序的数据,比如只输入payEvent不输入receiptEvent等,使之触发timer。

输入订单事件

nc -lk 8888

输入交易事件

nc -lk 9999

控制台输出:

success> 订单事件:OrderEvents{userId=1, action='pay', orId='orderId01', timestamp=1606555301}和交易事件:ReceiptEvents{orId='orderId01', payEquipment='wechat', timestamp=1606555307} 属于正常交易

好了今天的分享就到这,谢谢大家的阅读,如有疑问请留言或扫描下面二维码加笔者微信,一起交流。

扫描二维码

加笔者微信

获得更多精彩

Flink双流实时对账

本文分享自微信公众号 - OutOfMemoryError(backend_technology)。
如有侵权,请联系 support@oschina.cn 删除。
本文参与“OSC源创计划”,欢迎正在阅读的你也加入,一起分享。

点赞
收藏
评论区
推荐文章
blmius blmius
2年前
MySQL:[Err] 1292 - Incorrect datetime value: ‘0000-00-00 00:00:00‘ for column ‘CREATE_TIME‘ at row 1
文章目录问题用navicat导入数据时,报错:原因这是因为当前的MySQL不支持datetime为0的情况。解决修改sql\mode:sql\mode:SQLMode定义了MySQL应支持的SQL语法、数据校验等,这样可以更容易地在不同的环境中使用MySQL。全局s
Jacquelyn38 Jacquelyn38
2年前
2020年前端实用代码段,为你的工作保驾护航
有空的时候,自己总结了几个代码段,在开发中也经常使用,谢谢。1、使用解构获取json数据let jsonData  id: 1,status: "OK",data: 'a', 'b';let  id, status, data: number   jsonData;console.log(id, status, number )
Easter79 Easter79
1年前
thinkcmf+jsapi 实现微信支付
首先从小程序端接收订单号、金额等参数,然后后台进行统一下单,把微信支付的订单号返回,在把订单号发送给前台,前台拉起支付,返回参数后更改支付状态。。。回调publicfunctionnotify(){$wechatDb::name('wechat')where('status',1)find();
皕杰报表之UUID
​在我们用皕杰报表工具设计填报报表时,如何在新增行里自动增加id呢?能新增整数排序id吗?目前可以在新增行里自动增加id,但只能用uuid函数增加UUID编码,不能新增整数排序id。uuid函数说明:获取一个UUID,可以在填报表中用来创建数据ID语法:uuid()或uuid(sep)参数说明:sep布尔值,生成的uuid中是否包含分隔符'',缺省为
Stella981 Stella981
1年前
SpringBoot整合Redis乱码原因及解决方案
问题描述:springboot使用springdataredis存储数据时乱码rediskey/value出现\\xAC\\xED\\x00\\x05t\\x00\\x05问题分析:查看RedisTemplate类!(https://oscimg.oschina.net/oscnet/0a85565fa
Easter79 Easter79
1年前
Twitter的分布式自增ID算法snowflake (Java版)
概述分布式系统中,有一些需要使用全局唯一ID的场景,这种时候为了防止ID冲突可以使用36位的UUID,但是UUID有一些缺点,首先他相对比较长,另外UUID一般是无序的。有些时候我们希望能使用一种简单一些的ID,并且希望ID能够按照时间有序生成。而twitter的snowflake解决了这种需求,最初Twitter把存储系统从MySQL迁移
Easter79 Easter79
1年前
SpringBoot整合Redis乱码原因及解决方案
问题描述:springboot使用springdataredis存储数据时乱码rediskey/value出现\\xAC\\xED\\x00\\x05t\\x00\\x05问题分析:查看RedisTemplate类!(https://oscimg.oschina.net/oscnet/0a85565fa
Stella981 Stella981
1年前
RabbitMQ实现即时通讯居然如此简单!连后端代码都省得写了?
摘要有时候我们的项目中会用到即时通讯功能,比如电商系统中的客服聊天功能,还有在支付过程中,当用户支付成功后,第三方支付服务会回调我们的回调接口,此时我们需要通知前端支付成功。最近发现RabbitMQ可以很方便的实现即时通讯功能,如果你没有特殊的业务需求,甚至可以不写后端代码,今天给大家讲讲如何使用RabbitMQ来实现即时通讯!MQ
Wesley13 Wesley13
1年前
MySQL部分从库上面因为大量的临时表tmp_table造成慢查询
背景描述Time:20190124T00:08:14.70572408:00User@Host:@Id:Schema:sentrymetaLast_errno:0Killed:0Query_time:0.315758Lock_
鲍忠 鲍忠
10个月前
七月在线深度学习集训第三期一片孤城万仞山
七月在线深度学习集训第三期一片孤城万仞山如何关闭到期订单?在电子商务、支付等系统中,一般先创建一个订单(付款单),然后给用户一定的时间进行支付。如果没有按时付款,之前的订单(付款单)需要取消。类似的场景还有很多,到期自动收货,超时自动退款,