关于hadoop:6Flink-CEP-SQL模拟账号短时间内异地登录风控预警

本篇文章咱们来模仿一个实在的危险辨认场景，模仿 XX 平台上可能呈现盗号行为。技术实现计划：（1）通过将 xxx 平台用户登录时的登录日志发送到 kafka（本文代码演示用的 socket）；（2）Flink CEP SQL 规定引擎中定义好风控辨认规定，接入 kafka 数据源，比方一个账号在 5 分钟内，在多个不同地区有登录行为，那咱们认为该账号被盗；（3）Flink CEP 将辨认到的危险数据能够进行下发，为数据应用层提供数据服务，如：风控系统，数据大屏，态势感知 …..

（1）咱们先来定义一个数据生产者，模仿用户登录，产生登录日志：

package com.producers;

import java.io.BufferedWriter;
import java.io.IOException;
import java.io.OutputStreamWriter;
import java.net.ServerSocket;
import java.net.Socket;
import java.util.Random;

/**
 * Created by lj on 2022-08-10.
 */
public class Socket_Producer1 {public static void main(String[] args) throws IOException {

        try {ServerSocket ss = new ServerSocket(9999);
            System.out.println("启动 server ....");
            Socket s = ss.accept();
            BufferedWriter bw = new BufferedWriter(new OutputStreamWriter(s.getOutputStream()));
            String response = "java,1,2";

            // 每 2s 发送一次音讯
            int i = 0;
            Random r=new Random();   
            String[] userArr = {"user1","user2","user3","user4","user5","user6","user7","user8","user9"};
            String[] loginIP = {"167.234.67.123","219.141.178.14","220.180.239.202","111.73.240.192","123.182.253.242"};

            while(true){Thread.sleep(2000);
                response= userArr[r.nextInt(userArr.length)] + "," + loginIP[r.nextInt(loginIP.length)] +"\n";
                System.out.println(response);
                try{bw.write(response);
                    bw.flush();
                    i++;
                }catch (Exception ex){System.out.println(ex.getMessage());
                }

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

（2）在 CEP 中接入日志数据、定义风控规定

package com.examples;

import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.TableResult;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

import java.time.LocalDateTime;

import static org.apache.flink.table.api.Expressions.$;

/**
 * Created by lj on 2022-08-10.
 */
public class CEPSQLSocket1 {public static void main(String[] args) throws Exception {StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);
        DataStreamSource<String> streamSource = env.socketTextStream("127.0.0.1", 9999,"\n");
        SingleOutputStreamOperator<UserLoginLog> userLoginLog = streamSource.map(new MapFunction<String, UserLoginLog>() {
            @Override
            public UserLoginLog map(String s) throws Exception {String[] split = s.split(",");
                return new UserLoginLog(split[0], split[1], LocalDateTime.now());
            }
        });

        // 将流转化为表
        Table table = tableEnv.fromDataStream(userLoginLog,
                $("username"),
                $("ip"),
                $("rowtime1"),   //.rowtime()
                $("pt").proctime());

        CEP_SQL(env,tableEnv,table);
        env.execute();}

    private static void CEP_SQL(StreamExecutionEnvironment env,StreamTableEnvironment tEnv,Table table){System.out.println("===============CEP_SQL=================");

        tEnv.createTemporaryView("CEP_SQL", table);

        String sql = "SELECT *" +
                "FROM CEP_SQL" +
                "MATCH_RECOGNIZE (" +
                "PARTITION BY username" +
                "ORDER BY pt" +          // 在窗口内，对事件工夫进行排序。"MEASURES" +                   // 定义如何依据匹配胜利的输出事件结构输入事件
                "e1.username as user1,"+
                "First(e1.ip) as first_ip," +
                "LAST(e2.ip) as last_ip," +
                "e1.rowtime1 as rt," +
                "LAST(e2.pt) as end_tstamp" +           // 最新的事件工夫为 end_timestamp
                "ONE ROW PER MATCH" +                                      // 匹配胜利输入一条
                "AFTER MATCH  skip to next row" +                   // 匹配后跳转到下一行
                "PATTERN (e1 e2) WITHIN INTERVAL'5'MINUTE" +
                "DEFINE" +                                                 // 定义在 PATTERN 中呈现的 patternVariable 的具体含意
                "e1 AS" +
                "e1.username <>'', "+"            e2 AS "+"                e1.username = e2.username AND e1.ip <> e2.ip "+"    ) MR";


        TableResult res = tEnv.executeSql(sql);
//        while (res.collect().hasNext()){//            Row next = res.collect().next();
//            System.out.println(next);
//        }

        res.print();

        tEnv.dropTemporaryView("CEP_SQL");
    }

    public static class UserLoginLog {
        public  String username;
        public  String ip;
        public LocalDateTime rowtime1;

        public UserLoginLog(){}

        public UserLoginLog(String username,String ip,LocalDateTime rowtime){
            this.username = username;
            this.ip = ip;
            this.rowtime1 = rowtime;
        }

    }
}

（3）启动数据生产者，每 2 秒模仿一次用户登录行为

（4）启动 CEP 规定引擎服务，实时显示呈现异地登录的用户信息：