Flink基于 DataStream API 实现欺诈检测

浅浅的花香味﹌ 2022-11-15 11:19 282阅读 0赞

**目录**

系列文章目录

文章目录

前言

一、Flink基于 DataStream API 实现欺诈检测

二、使用步骤

1.引入pom.xml

2.主类

3.欺诈逻辑判断类

4.运行结果：

总结

--------------------

## 前言 ##

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

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

## 一、Flink基于 DataStream API 实现欺诈检测 ##

Flink基于 DataStream API 实现欺诈检测

## 二、使用步骤 ##

### 1.引入pom.xml ###

代码如下（示例）：

<?xml version="1.0" encoding="UTF-8"?>
    <project xmlns="http://maven.apache.org/POM/4.0.0"
             xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
             xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
        <parent>
            <artifactId>flinkdemo</artifactId>
            <groupId>com.flink.pangsl</groupId>
            <version>1.0-SNAPSHOT</version>
    
        </parent>
        <modelVersion>4.0.0</modelVersion>
    
        <artifactId>frauddetection</artifactId>
        <packaging>jar</packaging>
    
        <name>Flink Walkthrough DataStram Scala</name>
        <url>https://flink.apache.org</url>
    
        <repositories>
            <repository>
                <id>apache.snapshots</id>
                <name>Apache Development Snapshot Repository</name>
                <url>https://repository.apache.org/content/repositories/snapshots/</url>
                <releases>
                    <enabled>false</enabled>
                </releases>
                <snapshots>
                    <enabled>true</enabled>
                </snapshots>
            </repository>
        </repositories>
    
        <properties>
            <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
            <flink.version>1.12.0</flink.version>
            <scala.binary.version>2.11</scala.binary.version>
            <scala.version>2.11.12</scala.version>
            <log4j.version>2.12.1</log4j.version>
        </properties>
    
        <dependencies>
            <dependency>
                <groupId>org.apache.flink</groupId>
                <artifactId>flink-walkthrough-common_${scala.binary.version}</artifactId>
                <version>${flink.version}</version>
            </dependency>
    
            
            
            <dependency>
                <groupId>org.apache.flink</groupId>
                <artifactId>flink-streaming-scala_${scala.binary.version}</artifactId>
                <version>${flink.version}</version>
                <scope>compile</scope>
            </dependency>
            <dependency>
                <groupId>org.apache.flink</groupId>
                <artifactId>flink-clients_${scala.binary.version}</artifactId>
                <version>${flink.version}</version>
                <scope>compile</scope>
            </dependency>
    
            
    
            
    
            
            
            <dependency>
                <groupId>org.apache.logging.log4j</groupId>
                <artifactId>log4j-slf4j-impl</artifactId>
                <version>${log4j.version}</version>
                <scope>runtime</scope>
            </dependency>
            <dependency>
                <groupId>org.apache.logging.log4j</groupId>
                <artifactId>log4j-api</artifactId>
                <version>${log4j.version}</version>
                <scope>runtime</scope>
            </dependency>
            <dependency>
                <groupId>org.apache.logging.log4j</groupId>
                <artifactId>log4j-core</artifactId>
                <version>${log4j.version}</version>
                <scope>runtime</scope>
            </dependency>
        </dependencies>
    
        <build>
            <plugins>
                
                
                <plugin>
                    <groupId>org.apache.maven.plugins</groupId>
                    <artifactId>maven-shade-plugin</artifactId>
                    <version>3.0.0</version>
                    <executions>
                        
                        <execution>
                            <phase>package</phase>
                            <goals>
                                <goal>shade</goal>
                            </goals>
                            <configuration>
                                <artifactSet>
                                    <excludes>
                                        <exclude>org.apache.flink:force-shading</exclude>
                                        <exclude>com.google.code.findbugs:jsr305</exclude>
                                        <exclude>org.slf4j:*</exclude>
                                        <exclude>org.apache.logging.log4j:*</exclude>
                                    </excludes>
                                </artifactSet>
                                <filters>
                                    <filter>
                                        
                                        <artifact>*:*</artifact>
                                        <excludes>
                                            <exclude>META-INF/*.SF</exclude>
                                            <exclude>META-INF/*.DSA</exclude>
                                            <exclude>META-INF/*.RSA</exclude>
                                        </excludes>
                                    </filter>
                                </filters>
                                <transformers>
                                    <transformer implementation="org.apache.maven.plugins.shade.resource.ManifestResourceTransformer">
                                        <mainClass>spendreport.FraudDetectionJob</mainClass>
                                    </transformer>
                                </transformers>
                            </configuration>
                        </execution>
                    </executions>
                </plugin>
    
                
                <plugin>
                    <groupId>org.apache.maven.plugins</groupId>
                    <artifactId>maven-compiler-plugin</artifactId>
                    <version>3.1</version>
                    <configuration>
                        <source>1.8</source>
                        <target>1.8</target>
                    </configuration>
                </plugin>
    
                
                <plugin>
                    <groupId>net.alchim31.maven</groupId>
                    <artifactId>scala-maven-plugin</artifactId>
                    <version>3.2.2</version>
                    <executions>
                        <execution>
                            <goals>
                                <goal>compile</goal>
                                <goal>testCompile</goal>
                            </goals>
                        </execution>
                    </executions>
                    <configuration>
                        <args>
                            <arg>-nobootcp</arg>
                        </args>
                    </configuration>
                </plugin>
    
                
                <plugin>
                    <groupId>org.apache.maven.plugins</groupId>
                    <artifactId>maven-eclipse-plugin</artifactId>
                    <version>2.8</version>
                    <configuration>
                        <downloadSources>true</downloadSources>
                        <projectnatures>
                            <projectnature>org.scala-ide.sdt.core.scalanature</projectnature>
                            <projectnature>org.eclipse.jdt.core.javanature</projectnature>
                        </projectnatures>
                        <buildcommands>
                            <buildcommand>org.scala-ide.sdt.core.scalabuilder</buildcommand>
                        </buildcommands>
                        <classpathContainers>
                            <classpathContainer>org.scala-ide.sdt.launching.SCALA_CONTAINER</classpathContainer>
                            <classpathContainer>org.eclipse.jdt.launching.JRE_CONTAINER</classpathContainer>
                        </classpathContainers>
                        <excludes>
                            <exclude>org.scala-lang:scala-library</exclude>
                            <exclude>org.scala-lang:scala-compiler</exclude>
                        </excludes>
                        <sourceIncludes>
                            <sourceInclude>**/*.scala</sourceInclude>
                            <sourceInclude>**/*.java</sourceInclude>
                        </sourceIncludes>
                    </configuration>
                </plugin>
                <plugin>
                    <groupId>org.codehaus.mojo</groupId>
                    <artifactId>build-helper-maven-plugin</artifactId>
                    <version>1.9.1</version>
                    <executions>
                        
                        <execution>
                            <id>add-source</id>
                            <phase>generate-sources</phase>
                            <goals>
                                <goal>add-source</goal>
                            </goals>
                            <configuration>
                                <sources>
                                    <source>src/main/scala</source>
                                </sources>
                            </configuration>
                        </execution>
                        
                        <execution>
                            <id>add-test-source</id>
                            <phase>generate-test-sources</phase>
                            <goals>
                                <goal>add-test-source</goal>
                            </goals>
                            <configuration>
                                <sources>
                                    <source>src/test/scala</source>
                                </sources>
                            </configuration>
                        </execution>
                    </executions>
                </plugin>
            </plugins>
        </build>
    
    </project>

###    2.主类 ###

代码如下（示例）：

### 3.欺诈逻辑判断类 ###

/*
     * Licensed to the Apache Software Foundation (ASF) under one
     * or more contributor license agreements.  See the NOTICE file
     * distributed with this work for additional information
     * regarding copyright ownership.  The ASF licenses this file
     * to you under the Apache License, Version 2.0 (the
     * "License"); you may not use this file except in compliance
     * with the License.  You may obtain a copy of the License at
     *
     *     http://www.apache.org/licenses/LICENSE-2.0
     *
     * Unless required by applicable law or agreed to in writing, software
     * distributed under the License is distributed on an "AS IS" BASIS,
     * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
     * See the License for the specific language governing permissions and
     * limitations under the License.
     */
    
    package spendreport
    
    import org.apache.flink.api.common.state.{ValueState, ValueStateDescriptor}
    import org.apache.flink.api.scala.typeutils.Types
    import org.apache.flink.configuration.Configuration
    import org.apache.flink.streaming.api.functions.KeyedProcessFunction
    import org.apache.flink.util.Collector
    import org.apache.flink.walkthrough.common.entity.Alert
    import org.apache.flink.walkthrough.common.entity.Transaction
    
    /**
      * Skeleton code for implementing a fraud detector.
      */
    object FraudDetector {
      val SMALL_AMOUNT: Double = 1.00
      val LARGE_AMOUNT: Double = 500.00
      val ONE_MINUTE: Long     = 60 * 1000L
    }
    
    @SerialVersionUID(1L)
    class FraudDetector extends KeyedProcessFunction[Long, Transaction, Alert] {
    
      @transient private var flagState: ValueState[java.lang.Boolean] = _
      @transient private var timerState: ValueState[java.lang.Long] = _
    
      @throws[Exception]
      override def open(parameters: Configuration): Unit = {
        val flagDescriptor = new ValueStateDescriptor("flag", Types.BOOLEAN)
        flagState = getRuntimeContext.getState(flagDescriptor)
    
        val timerDescriptor = new ValueStateDescriptor("timer-state", Types.LONG)
        timerState = getRuntimeContext.getState(timerDescriptor)
      }
    
      override def processElement(
                                   transaction: Transaction,
                                   context: KeyedProcessFunction[Long, Transaction, Alert]#Context,
                                   collector: Collector[Alert]): Unit = {
    
        // Get the current state for the current key
        val lastTransactionWasSmall = flagState.value
    
        // Check if the flag is set
        if (lastTransactionWasSmall != null) {
          if (transaction.getAmount > FraudDetector.LARGE_AMOUNT) {
            // Output an alert downstream
            val alert = new Alert
            alert.setId(transaction.getAccountId)
    
            collector.collect(alert)
          }
          // Clean up our state
          cleanUp(context)
        }
    
        if (transaction.getAmount < FraudDetector.SMALL_AMOUNT) {
          // set the flag to true
          flagState.update(true)
          val timer = context.timerService.currentProcessingTime + FraudDetector.ONE_MINUTE
    
          context.timerService.registerProcessingTimeTimer(timer)
          timerState.update(timer)
        }
      }
    
      override def onTimer(
                            timestamp: Long,
                            ctx: KeyedProcessFunction[Long, Transaction, Alert]#OnTimerContext,
                            out: Collector[Alert]): Unit = {
        // remove flag after 1 minute
        timerState.clear()
        flagState.clear()
      }
    
      @throws[Exception]
      private def cleanUp(ctx: KeyedProcessFunction[Long, Transaction, Alert]#Context): Unit = {
        // delete timer
        val timer = timerState.value
        ctx.timerService.deleteProcessingTimeTimer(timer)
    
        // clean up all states
        timerState.clear()
        flagState.clear()
      }
    }

--------------------

### 4.运行结果： ###

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMTExMTAzMDE_size_16_color_FFFFFF_t_70][]

## 总结 ##

以上就是今天要讲的内容，本文介绍了基于 DataStream API 实现欺诈检测。

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMTExMTAzMDE_size_16_color_FFFFFF_t_70]: /images/20221022/a08d55b036d043508b39af856ac3469a.png