image.png
本篇文章我們來(lái)模擬一個(gè)真實(shí)的風(fēng)險(xiǎn)識(shí)別場(chǎng)景，模擬XX平臺(tái)上可能出現(xiàn)盜號(hào)行為毙玻。
技術(shù)實(shí)現(xiàn)方案：
（1）通過將xxx平臺(tái)用戶登錄時(shí)的登錄日志發(fā)送到kafka（本文代碼演示用的socket）失驶；
（2）Flink CEP SQL規(guī)則引擎中定義好風(fēng)控識(shí)別規(guī)則勺届，接入kafka數(shù)據(jù)源驶俊，比如一個(gè)賬號(hào)在5分鐘內(nèi)，在多個(gè)不同地區(qū)有登錄行為免姿，那我們認(rèn)為該賬號(hào)被盜饼酿；
（3）Flink CEP將識(shí)別到的風(fēng)險(xiǎn)數(shù)據(jù)可以進(jìn)行下發(fā)，為數(shù)據(jù)應(yīng)用層提供數(shù)據(jù)服務(wù)胚膊，如：風(fēng)控系統(tǒng)故俐，數(shù)據(jù)大屏，態(tài)勢(shì)感知.....
image.png
（1）我們先來(lái)定義一個(gè)數(shù)據(jù)生產(chǎn)者紊婉，模擬用戶登錄药版，產(chǎn)生登錄日志：
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("啟動(dòng) server ....");
Socket s = ss.accept();
BufferedWriter bw = new BufferedWriter(new OutputStreamWriter(s.getOutputStream()));
String response = "java,1,2";
//每 2s 發(fā)送一次消息
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中接入日志數(shù)據(jù)、定義風(fēng)控規(guī)則
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());
}
});
// 將流轉(zhuǎn)化為表
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 " + //在窗口內(nèi)喻犁，對(duì)事件時(shí)間進(jìn)行排序槽片。
" MEASURES " + //定義如何根據(jù)匹配成功的輸入事件構(gòu)造輸出事件
" 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 " + //最新的事件時(shí)間為end_timestamp
" ONE ROW PER MATCH " + //匹配成功輸出一條
" AFTER MATCH skip to next row " + //匹配后跳轉(zhuǎn)到下一行
" PATTERN ( e1 e2 ) WITHIN INTERVAL '5' MINUTE " +
" DEFINE " + //定義在PATTERN中出現(xiàn)的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）啟動(dòng)數(shù)據(jù)生產(chǎn)者，每2秒模擬一次用戶登錄行為
image.png
（4）啟動(dòng)CEP規(guī)則引擎服務(wù)肢础，實(shí)時(shí)顯示出現(xiàn)異地登錄的用戶信息：
image.png
