学习笔记Flink(八)—— 基于Flink 在线交易反欺诈检测
一、背景介紹
信用卡欺詐
信用卡欺詐是指故意使用偽造、作廢的信用卡,冒用他人的信用卡騙取財(cái)物,或用本人信用卡進(jìn)行惡意透支的行為。在當(dāng)今數(shù)字時(shí)代,信用卡欺詐行為越來(lái)越被重視。
罪犯可以通過(guò)詐騙或者入侵安全級(jí)別較低系統(tǒng)來(lái)盜竊信用卡卡號(hào)。 用盜得的信用卡進(jìn)行很小額度的消費(fèi)進(jìn)行測(cè)試。 如果測(cè)試消費(fèi)成功,那么他們就會(huì)用這個(gè)信用卡進(jìn)行大筆消費(fèi)。
信用卡欺詐行為
交易3和交易4應(yīng)該被標(biāo)記為欺詐行為,因?yàn)榻灰?是一個(gè)100¥的小額交易,而緊隨著的交易4是一個(gè)10000¥的大額交易。
另外,交易5、6和交易7就不屬于欺詐交易了,因?yàn)樵诮灰?這個(gè)500¥的小額交易之后,并沒(méi)有跟隨一個(gè)大額交易,而是一個(gè)金額適中的交易,這使得交易5到交易7不屬于欺詐行為。
二、架構(gòu)設(shè)計(jì)
架構(gòu)設(shè)計(jì)
數(shù)據(jù)流設(shè)計(jì)
數(shù)據(jù)流落地實(shí)現(xiàn)
三、Kafka信用卡消費(fèi)數(shù)據(jù)
3.1、Kafka Producer
模擬Kafka Producer定時(shí)生成消費(fèi)數(shù)據(jù)
TransactionData.java:
package fraud_detection;public class TransactionData {private String user;private double money;public TransactionData(){}public TransactionData(String user,double money){this.user=user;this.money=money;}@Overridepublic String toString(){return this.user + "," + this.money;} }TransactionDataGenerator.java:
package fraud_detection;import java.util.Random;public class TransactionDataGenerator {public static final int USER_SIZE = 10;public static final float BIG_MONEY_PERCENT = 0.02f;static Random random = new Random();public static TransactionData getData(){return new TransactionData(generateUser() , generateMoney()) ;}private static String generateUser(){return "user_"+random.nextInt(USER_SIZE);}private static float generateMoney(){float i = random.nextFloat();if( i > BIG_MONEY_PERCENT){return random.nextFloat() * 1000;}else{return i * 10000000;}}public static void main(String[] args){TransactionData data = null;for(int i = 10000 ;i >0 ; i--){data = getData();System.out.println(data);}} }TransactionDataProducer.java:
package fraud_detection; import org.apache.kafka.clients.producer.KafkaProducer; import org.apache.kafka.clients.producer.Producer; import org.apache.kafka.clients.producer.ProducerRecord; import java.util.HashMap; import java.util.Map; public class TransactionDataProducer {public static void main(String[] args) throws InterruptedException {String topic = "fraud00";Map<String,Object> kafkaProperties = new HashMap<>();kafkaProperties.put("bootstrap.servers","node100:9092,node101:9092,node102:9092");kafkaProperties.put("acks", "all");kafkaProperties.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");kafkaProperties.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");Producer<String, String> producer = new KafkaProducer<>(kafkaProperties);int size = 30*1000/10;long interval = 10L;String data = "";for (int i = 0; i < size; i++) {Thread.sleep(interval);data= TransactionDataGenerator.getData().toString();producer.send(new ProducerRecord<>(topic, data));}producer.close();System.out.println("消息發(fā)送完成!");} }3.2、整合Kafka Transaction數(shù)據(jù)
FraudDetection.scala:
package flink_kafka import java.util.Properties import org.apache.flink.api.common.serialization.SimpleStringSchema import org.apache.flink.streaming.api.scala._ import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer object FraudDetection {def main(args: Array[String]): Unit = {val topic = "fraud00"val properties = new Properties()properties.setProperty("bootstrap.servers", "node100:9092")properties.setProperty("group.id", "test")val env = StreamExecutionEnvironment.getExecutionEnvironmentval stream = env.addSource(new FlinkKafkaConsumer[String](topic, new SimpleStringSchema(), properties))stream.map(x => {val data = x.split(","); (data(0).trim, data(1).trim.toDouble)}).keyBy(0).mapWithState[(String, Double), Double]((in: (String, Double), state: Option[Double]) =>state match {case None => (("", 0.0), Some(in._2))case Some(previous) => if (in._2 > 10000.0 && previous < 1000.0)((in._1 + "->" + previous, in._2), Some(in._2)) else (("", 0.0), Some(in._2))}).filter(x => x._2 > 0.0).print()env.execute("Fraud Detection")} }測(cè)試:
① 先創(chuàng)建fraud00 話題
將產(chǎn)生的數(shù)據(jù)存到/tmp目錄下(了解)
② 運(yùn)行FraudDetection:
③ 運(yùn)行TransactionDataProducer
結(jié)果:
總結(jié)
以上是生活随笔為你收集整理的学习笔记Flink(八)—— 基于Flink 在线交易反欺诈检测的全部?jī)?nèi)容,希望文章能夠幫你解決所遇到的問(wèn)題。
- 上一篇: 学习笔记Flink(七)—— Flink
- 下一篇: XML属性列表