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

案例背景

在当今数字时代，信用卡欺诈行为越来越被重视。 罪犯可以通过诈骗或者入侵安全级别较低系统来盗窃信用卡卡号。 用盗得的信用卡进行很小额度的例如一美元或者更小额度的消费进行测试。 如果测试消费成功，那么他们就会用这个信用卡进行大笔消费，来购买一些他们希望得到的，或者可以倒卖的财物。

在这个教程中，你将会建立一个针对可疑信用卡交易行为的反欺诈检测系统。 通过使用一组简单的规则，你将了解到 Flink 如何为我们实现复杂业务逻辑并实时执行。

欺诈检测规则

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

对原有案例进行改造

1. 数据源使用Kafka，发送json格式字符串
消息格式:  {"accountId":1001, "timestamp":1656490723171, "amount":0.12}

2. 自定义 DeserializationSchema, 直接将kafka的json字符串转成POJO对象

流程图

核心代码

1. 自定义DeserializationSchema

public class TransactionDeserialization implements DeserializationSchema<Transaction> {
    @Override
    public Transaction deserialize(byte[] bytes) throws IOException {
        ByteBuffer buffer = ByteBuffer.wrap(bytes);
        String message = byteBufferToString(buffer);
        if (StringUtils.isBlank(message)) {
            return null;
        }
        Transaction transaction = JsonUtils.fromJson(message, Transaction.class);
        return transaction;
    }

    @Override
    public boolean isEndOfStream(Transaction transaction) {
        return false;
    }

    @Override
    public TypeInformation<Transaction> getProducedType() {
        return TypeInformation.of(Transaction.class);
    }



    /**
     * ByteBuffer 转换 String
     * @param buffer
     * @return
     */
    private String byteBufferToString(ByteBuffer buffer) {
        String ret = "";
        try{
            CharsetDecoder decoder = Charset.forName("UTF-8").newDecoder();
            CharBuffer charBuffer = decoder.decode(buffer.asReadOnlyBuffer());;
            ret = charBuffer.toString();
        }catch (Exception e) {
            e.printStackTrace();
        }
        return ret;
    }
}

2. 欺诈检测核心代码

public class FraudDetector extends KeyedProcessFunction<Long, Transaction, Alert> {

    /**
     * 定义小金额边界
     */
    private static final double SMALL_AMOUNT = 1.00;

    /**
     * 定义大金额边界
     */
    private static final double LARGE_AMOUNT = 500.00;

    /**
     * 1分钟时间
     */
    private static final long ONE_MINUTE = 60 * 1000;

    /**
     * 保存是否有消费小金额的状态
     */
    private transient ValueState<Boolean> smallAmountState;

    /**
     * 定时器状态
     */
    private transient ValueState<Long> timerState;

    @Override
    public void open(Configuration parameters) throws Exception {
        // 初始化ValueState

        ValueStateDescriptor<Boolean> smallAmountStateDescriptor = new ValueStateDescriptor<Boolean>("small-amount-state", Types.BOOLEAN);
        smallAmountState = getRuntimeContext().getState(smallAmountStateDescriptor);

        ValueStateDescriptor<Long> timerStateDescriptor = new ValueStateDescriptor<Long>("timer-state", Types.LONG);
        timerState = getRuntimeContext().getState(timerStateDescriptor);

    }

    @Override
    public void processElement(Transaction transaction, Context context, Collector<Alert> collector) throws Exception {
        if (Objects.isNull(transaction)) {
            return;
        }
        // Get the current state for the current key
        Boolean lastTransactionWasSmall = smallAmountState.value();

        // Check if the flag is set
        if (Objects.nonNull(lastTransactionWasSmall)) {
            if (transaction.getAmount() > LARGE_AMOUNT) {
                Alert alert = new Alert();
                alert.setAccountId(transaction.getAccountId());
                alert.setAmount(transaction.getAmount());

                collector.collect(alert);
            }
            clearUp(context);
        }

        if (transaction.getAmount() < SMALL_AMOUNT) {
            // set the flag to true
            smallAmountState.update(true);

            // 注册定时器，设置一个当前时间一分钟后触发的定时器，同时，将触发时间保存到 timerState 状态中。
            long timer = context.timerService().currentProcessingTime() + ONE_MINUTE;
            context.timerService().registerProcessingTimeTimer(timer);
            timerState.update(timer);
        }

    }

    @Override
    public void onTimer(long timestamp, OnTimerContext ctx, Collector<Alert> out) throws Exception {
        // remove flag after 1 minute
        timerState.clear();
        smallAmountState.clear();
    }

    private void clearUp(Context ctx) {
        try {
            // delete timer
            Long timer = timerState.value();
            ctx.timerService().deleteProcessingTimeTimer(timer);

            timerState.clear();
            smallAmountState.clear();

        } catch (IOException e) {
            e.printStackTrace();
        }
    }
}

3. FLink Job 启动类

public class FraudDetectionJob {
    public static void main(String[] args) throws Exception {
        // 初始化环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        // kafka消息格式: {"accountId":1001, "timestamp":1656490723171, "amount":0.12}

        // 定义Kafka数据源
        KafkaSource<Transaction> source = KafkaSource.<Transaction>builder()
                .setBootstrapServers("192.168.0.192:9092")
                .setTopics("TOPIC_FRAUD_DETECTION")
                .setGroupId("TEST_GROUP")
                .setStartingOffsets(OffsetsInitializer.latest())
                .setValueOnlyDeserializer(new TransactionDeserialization())
                .build();

        // 加载数据源
        DataStreamSource<Transaction> fraudDetectionSource
                = env.fromSource(source, WatermarkStrategy.noWatermarks(), "FraudDetection-Source");

        // 处理数据
        SingleOutputStreamOperator<Alert> alertStreamOperator = fraudDetectionSource.keyBy(Transaction::getAccountId)
                .process(new FraudDetector())
                .name("Fraud-Detector");

        // 输出告警结果
        alertStreamOperator.addSink(new AlertSink())
                .name("Send-Alerts");

        env.execute("Fraud Detection");

    }
}

执行效果

1. kafka输入
   

2. 告警结果
   

完整代码

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

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