Flink-基于 DataStream API 实现欺诈检测

案例来源于 https://nightlies.apache.org/flink/flink-docs-release-1.14/docs/try-flink/datastream/

案例背景

在当今数字时代，信用卡欺诈行为越来越被重视。 罪犯可以通过诈骗或者入侵安全级别较低系统来盗窃信用卡卡号。 用盗得的信用卡进行很小额度的例如一美元或者更小额度的消费进行测试。 如果测试消费成功，那么他们就会用这个信用卡进行大笔消费，来购买一些他们希望得到的，或者可以倒卖的财物。
在这个教程中，你将会建立一个针对可疑信用卡交易行为的反欺诈检测系统。 通过使用一组简单的规则，你将了解到 Flink 如何为我们实现复杂业务逻辑并实时执行。
欺诈检测规则

• 对于一个账户，如果出现一笔小于1元的交易后, 紧跟着在1分钟内又出现一笔大于500元的交易，则认为该账户属于欺诈，就输出一个报警消息。
  
• 图说明如下
  
  

对原有案例进行改造

1. 1. 数据源使用Kafka，发送json格式字符串
2. 消息格式:  {"accountId":1001, "timestamp":1656490723171, "amount":0.12}
4. 2. 自定义 DeserializationSchema, 直接将kafka的json字符串转成POJO对象

复制代码

流程图


核心代码

• 自定义DeserializationSchema
  

1. public class TransactionDeserialization implements DeserializationSchema<Transaction> {
2.     @Override
3.     public Transaction deserialize(byte[] bytes) throws IOException {
4.         ByteBuffer buffer = ByteBuffer.wrap(bytes);
5.         String message = byteBufferToString(buffer);
6.         if (StringUtils.isBlank(message)) {
7.             return null;
8.         }
9.         Transaction transaction = JsonUtils.fromJson(message, Transaction.class);
10.         return transaction;
11.     }
13.     @Override
14.     public boolean isEndOfStream(Transaction transaction) {
15.         return false;
16.     }
18.     @Override
19.     public TypeInformation<Transaction> getProducedType() {
20.         return TypeInformation.of(Transaction.class);
21.     }
25.     /**
26.      * ByteBuffer 转换 String
27.      * @param buffer
28.      * @return
29.      */
30.     private String byteBufferToString(ByteBuffer buffer) {
31.         String ret = "";
32.         try{
33.             CharsetDecoder decoder = Charset.forName("UTF-8").newDecoder();
34.             CharBuffer charBuffer = decoder.decode(buffer.asReadOnlyBuffer());;
35.             ret = charBuffer.toString();
36.         }catch (Exception e) {
37.             e.printStackTrace();
38.         }
39.         return ret;
40.     }
41. }

复制代码

• 欺诈检测核心代码
  

1. public class FraudDetector extends KeyedProcessFunction<Long, Transaction, Alert> {
3.     /**
4.      * 定义小金额边界
5.      */
6.     private static final double SMALL_AMOUNT = 1.00;
8.     /**
9.      * 定义大金额边界
10.      */
11.     private static final double LARGE_AMOUNT = 500.00;
13.     /**
14.      * 1分钟时间
15.      */
16.     private static final long ONE_MINUTE = 60 * 1000;
18.     /**
19.      * 保存是否有消费小金额的状态
20.      */
21.     private transient ValueState<Boolean> smallAmountState;
23.     /**
24.      * 定时器状态
25.      */
26.     private transient ValueState<Long> timerState;
28.     @Override
29.     public void open(Configuration parameters) throws Exception {
30.         // 初始化ValueState
32.         ValueStateDescriptor<Boolean> smallAmountStateDescriptor = new ValueStateDescriptor<Boolean>("small-amount-state", Types.BOOLEAN);
33.         smallAmountState = getRuntimeContext().getState(smallAmountStateDescriptor);
35.         ValueStateDescriptor<Long> timerStateDescriptor = new ValueStateDescriptor<Long>("timer-state", Types.LONG);
36.         timerState = getRuntimeContext().getState(timerStateDescriptor);
38.     }
40.     @Override
41.     public void processElement(Transaction transaction, Context context, Collector<Alert> collector) throws Exception {
42.         if (Objects.isNull(transaction)) {
43.             return;
44.         }
45.         // Get the current state for the current key
46.         Boolean lastTransactionWasSmall = smallAmountState.value();
48.         // Check if the flag is set
49.         if (Objects.nonNull(lastTransactionWasSmall)) {
50.             if (transaction.getAmount() > LARGE_AMOUNT) {
51.                 Alert alert = new Alert();
52.                 alert.setAccountId(transaction.getAccountId());
53.                 alert.setAmount(transaction.getAmount());
55.                 collector.collect(alert);
56.             }
57.             clearUp(context);
58.         }
60.         if (transaction.getAmount() < SMALL_AMOUNT) {
61.             // set the flag to true
62.             smallAmountState.update(true);
64.             // 注册定时器，设置一个当前时间一分钟后触发的定时器，同时，将触发时间保存到 timerState 状态中。
65.             long timer = context.timerService().currentProcessingTime() + ONE_MINUTE;
66.             context.timerService().registerProcessingTimeTimer(timer);
67.             timerState.update(timer);
68.         }
70.     }
72.     @Override
73.     public void onTimer(long timestamp, OnTimerContext ctx, Collector<Alert> out) throws Exception {
74.         // remove flag after 1 minute
75.         timerState.clear();
76.         smallAmountState.clear();
77.     }
79.     private void clearUp(Context ctx) {
80.         try {
81.             // delete timer
82.             Long timer = timerState.value();
83.             ctx.timerService().deleteProcessingTimeTimer(timer);
85.             timerState.clear();
86.             smallAmountState.clear();
88.         } catch (IOException e) {
89.             e.printStackTrace();
90.         }
91.     }
92. }

复制代码

• FLink Job 启动类
  

1. public class FraudDetectionJob {
2.     public static void main(String[] args) throws Exception {
3.         // 初始化环境
4.         StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
6.         // kafka消息格式: {"accountId":1001, "timestamp":1656490723171, "amount":0.12}
8.         // 定义Kafka数据源
9.         KafkaSource<Transaction> source = KafkaSource.<Transaction>builder()
10.                 .setBootstrapServers("192.168.0.192:9092")
11.                 .setTopics("TOPIC_FRAUD_DETECTION")
12.                 .setGroupId("TEST_GROUP")
13.                 .setStartingOffsets(OffsetsInitializer.latest())
14.                 .setValueOnlyDeserializer(new TransactionDeserialization())
15.                 .build();
17.         // 加载数据源
18.         DataStreamSource<Transaction> fraudDetectionSource
19.                 = env.fromSource(source, WatermarkStrategy.noWatermarks(), "FraudDetection-Source");
21.         // 处理数据
22.         SingleOutputStreamOperator<Alert> alertStreamOperator = fraudDetectionSource.keyBy(Transaction::getAccountId)
23.                 .process(new FraudDetector())
24.                 .name("Fraud-Detector");
26.         // 输出告警结果
27.         alertStreamOperator.addSink(new AlertSink())
28.                 .name("Send-Alerts");
30.         env.execute("Fraud Detection");
32.     }
33. }

复制代码

执行效果

• kafka输入
  
  
• 告警结果
  
  

完整代码

https://github.com/Mr-LuXiaoHua/study-flink

1. 代码入口: com.example.datastream.frauddetection.FraudDetectionJob

复制代码

免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！