Flink实战二之使用 DataStream API 进行欺诈检测
·
前言
Apache Flink 提供了用于构建健壮的有状态流应用程序的 DataStream API。它提供了对状态和时间的细粒度控制，从而能够实现高级的事件驱动系统。在这里，你将学习如何使用 Flink 的 DataStream API 构建有状态流应用程序。
你将构建什么
在本教程中，你将构建一个欺诈检测系统，用于对可疑的信用卡交易发出警报。通过一套简单的规则，你将看到 Flink 如何让我们实现高级业务逻辑并实时采取行动。
前置条件
本教程假设你对 Java 有一定了解，但即使你使用其他编程语言，也应该能够顺利跟随。
在 IDE 中运行
在 IDE 中运行项目可能会出现 java.lang.NoClassDefFoundError 异常。这可能是因为你没有将所有必需的 Flink 依赖隐式加载到类路径中。
IntelliJ IDEA：依次点击 Run > Edit Configurations > Modify options > Select include dependencies with "Provided" scope。此运行配置现在将包含在 IDE 中运行应用程序所需的所有类。
如何跟随教程
如果你想跟随教程，你需要一台配备以下软件的电脑：
- Java 11
- Maven
我们提供的 Flink Maven 原型将快速创建一个包含所有必要依赖的骨架项目，因此你只需专注于填写业务逻辑即可。这些依赖包括：
flink-streaming-java：所有 Flink 流处理应用的核心依赖flink-walkthrough-common：包含本教程专用的数据生成器和其他类
yangyanping@yangyaningdeAir bin % ./mvn archetype:generate \
-DarchetypeGroupId=org.apache.flink \
-DarchetypeArtifactId=flink-walkthrough-datastream-java \
-DarchetypeVersion=2.2.0 \
-DgroupId=frauddetection \
-DartifactId=frauddetection \
-Dversion=0.1 \
-Dpackage=spendreport \
-DinteractiveMode=false
[INFO] Scanning for projects...
......
[INFO] ----------------------------------------------------------------------------
[INFO] Using following parameters for creating project from Archetype: flink-walkthrough-datastream-java:2.2.0
[INFO] ----------------------------------------------------------------------------
[INFO] Parameter: groupId, Value: frauddetection
[INFO] Parameter: artifactId, Value: frauddetection
[INFO] Parameter: version, Value: 0.1
[INFO] Parameter: package, Value: spendreport
[INFO] Parameter: packageInPathFormat, Value: spendreport
[INFO] Parameter: package, Value: spendreport
[INFO] Parameter: groupId, Value: frauddetection
[INFO] Parameter: artifactId, Value: frauddetection
[INFO] Parameter: version, Value: 0.1
[WARNING] CP Don't override file /Users/yangyanping/Downloads/apache-maven-3.5.4/bin/frauddetection/src/main/resources
[INFO] Project created from Archetype in dir: /Users/yangyanping/Downloads/apache-maven-3.5.4/bin/frauddetection
[INFO] ------------------------------------------------------------------------
[INFO] BUILD SUCCESS
[INFO] ------------------------------------------------------------------------
[INFO] Total time: 01:12 min
[INFO] Finished at: 2026-02-27T07:33:54+08:00
[INFO] ------------------------------------------------------------------------你可以根据需要编辑 groupId、artifactId 和 package。使用上述参数，Maven 将创建一个名为 frauddetection 的文件夹，其中包含一个带有完成本教程所需所有依赖的项目。
将项目导入编辑器后，你可以找到一个名为 FraudDetectionJob.java 的文件，其中包含以下代码，你可以直接在 IDE 中运行这些代码。尝试在整个数据流中设置断点，并在 DEBUG 模式下运行代码，以了解一切是如何运作的。
pom #
<!--
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.
-->
<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">
<modelVersion>4.0.0</modelVersion>
<groupId>frauddetection</groupId>
<artifactId>frauddetection</artifactId>
<version>0.1</version>
<packaging>jar</packaging>
<name>Flink Walkthrough DataStream Java</name>
<url>https://flink.apache.org</url>
<properties>
<project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
<flink.version>2.2.0</flink.version>
<target.java.version>1.8</target.java.version>
<maven.compiler.source>${target.java.version}</maven.compiler.source>
<maven.compiler.target>${target.java.version}</maven.compiler.target>
<log4j.version>2.24.3</log4j.version>
</properties>
<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>
<dependencies>
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-walkthrough-common</artifactId>
<version>${flink.version}</version>
</dependency>
<!-- This dependency is provided, because it should not be packaged into the JAR file. -->
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-streaming-java</artifactId>
<version>${flink.version}</version>
<scope>provided</scope>
</dependency>
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-clients</artifactId>
<version>${flink.version}</version>
<scope>provided</scope>
</dependency>
<!-- Add connector dependencies here. They must be in the default scope (compile). -->
<!-- Example:
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-connector-kafka</artifactId>
<version>3.0.0-1.17</version>
</dependency>
-->
<!-- Add logging framework, to produce console output when running in the IDE. -->
<!-- These dependencies are excluded from the application JAR by default. -->
<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>
<!-- Java Compiler -->
<plugin>
<groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-compiler-plugin</artifactId>
<version>3.1</version>
<configuration>
<source>${target.java.version}</source>
<target>${target.java.version}</target>
</configuration>
</plugin>
<!-- We use the maven-shade plugin to create a fat jar that contains all necessary dependencies. -->
<!-- Change the value of <mainClass>...</mainClass> if your program entry point changes. -->
<plugin>
<groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-shade-plugin</artifactId>
<version>3.0.0</version>
<executions>
<!-- Run shade goal on package phase -->
<execution>
<phase>package</phase>
<goals>
<goal>shade</goal>
</goals>
<configuration>
<artifactSet>
<excludes>
<exclude>org.apache.flink:flink-shaded-force-shading</exclude>
<exclude>com.google.code.findbugs:jsr305</exclude>
<exclude>org.slf4j:*</exclude>
<exclude>org.apache.logging.log4j:*</exclude>
</excludes>
</artifactSet>
<filters>
<filter>
<!-- Do not copy the signatures in the META-INF folder.
Otherwise, this might cause SecurityExceptions when using the JAR. -->
<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>
</plugins>
<pluginManagement>
<plugins>
<!-- This improves the out-of-the-box experience in Eclipse by resolving some warnings. -->
<plugin>
<groupId>org.eclipse.m2e</groupId>
<artifactId>lifecycle-mapping</artifactId>
<version>1.0.0</version>
<configuration>
<lifecycleMappingMetadata>
<pluginExecutions>
<pluginExecution>
<pluginExecutionFilter>
<groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-shade-plugin</artifactId>
<versionRange>[3.0.0,)</versionRange>
<goals>
<goal>shade</goal>
</goals>
</pluginExecutionFilter>
<action>
<ignore/>
</action>
</pluginExecution>
<pluginExecution>
<pluginExecutionFilter>
<groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-compiler-plugin</artifactId>
<versionRange>[3.1,)</versionRange>
<goals>
<goal>testCompile</goal>
<goal>compile</goal>
</goals>
</pluginExecutionFilter>
<action>
<ignore/>
</action>
</pluginExecution>
</pluginExecutions>
</lifecycleMappingMetadata>
</configuration>
</plugin>
</plugins>
</pluginManagement>
</build>
</project>
FraudDetectionJob.java #
/*
* 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.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.walkthrough.common.sink.AlertSink;
import org.apache.flink.walkthrough.common.entity.Alert;
import org.apache.flink.walkthrough.common.entity.Transaction;
import org.apache.flink.walkthrough.common.source.TransactionSource;
/**
* Skeleton code for the datastream walkthrough
*/
public class FraudDetectionJob {
public static void main(String[] args) throws Exception {
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
DataStream<Transaction> transactions = env
.addSource(new TransactionSource())
.name("transactions");
DataStream<Alert> alerts = transactions
.keyBy(Transaction::getAccountId)
.process(new FraudDetector())
.name("fraud-detector");
alerts
.addSink(new AlertSink())
.name("send-alerts");
env.execute("Fraud Detection");
}
}
FraudDetector.java #
/*
* 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.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.
*/
public class FraudDetector extends KeyedProcessFunction<Long, Transaction, Alert> {
private static final long serialVersionUID = 1L;
private static final double SMALL_AMOUNT = 1.00;
private static final double LARGE_AMOUNT = 500.00;
private static final long ONE_MINUTE = 60 * 1000;
@Override
public void processElement(
Transaction transaction,
Context context,
Collector<Alert> collector) throws Exception {
Alert alert = new Alert();
alert.setId(transaction.getAccountId());
collector.collect(alert);
}
}
代码解析
让我们逐步解析这两个文件的代码。FraudDetectionJob 类定义了应用程序的数据流，而 FraudDetector 类则定义了检测欺诈交易的函数的业务逻辑。
我们首先从 FraudDetectionJob 类的 main 方法开始，描述作业是如何组装的。
执行环境
第一行代码会设置你的 StreamExecutionEnvironment。执行环境是你为作业设置属性、创建数据源，并最终触发作业执行的地方。
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();创建数据源
数据源将来自外部系统（如 Apache Kafka、Rabbit MQ 或 Apache Pulsar）的数据摄入到 Flink 作业中。本教程使用一个生成无限信用卡交易流的数据源供你处理。每笔交易包含账户 ID（accountId）、交易发生的时间戳（timestamp）以及美元金额（amount）。附加在数据源上的名称仅用于调试目的，这样如果出现问题，我们就能知道错误源自哪里。
DataStream<Transaction> transactions = env
.addSource(new TransactionSource())
.name("transactions");事件分区与欺诈检测
交易流包含来自大量用户的众多交易，因此需要由多个欺诈检测任务并行处理。由于欺诈是按账户发生的，你必须确保同一账户的所有交易都由欺诈检测器的同一个并行任务处理。
为了确保同一任务处理特定键的所有记录，你可以使用 DataStream#keyBy 对流进行分区。调用 process() 会添加一个算子，该算子将函数应用于流中每个已分区的元素。通常认为，紧跟在 keyBy 之后的算子（在本例中为 FraudDetector）是在键控上下文中执行的。
DataStream<Alert> alerts = transactions
.keyBy(Transaction::getAccountId)
.process(new FraudDetector())
.name("fraud-detector");输出结果
接收器（Sink）将数据流写入外部系统，如 Apache Kafka、Cassandra 和 AWS Kinesis。警报接收器（AlertSink）会以 INFO 日志级别记录每条警报记录，而非将其写入持久化存储，这样你就能轻松查看结果。
alerts.addSink(new AlertSink());欺诈检测器
欺诈检测器被实现为一个 KeyedProcessFunction。它的 KeyedProcessFunction#processElement 方法会在每个交易事件发生时被调用。这个第一个版本会对每笔交易都产生警报，有些人可能会说这过于保守了。
本教程的后续步骤将指导你用更有意义的业务逻辑来扩展这个欺诈检测器。
public class FraudDetector extends KeyedProcessFunction<Long, Transaction, Alert> {
private static final double SMALL_AMOUNT = 1.00;
private static final double LARGE_AMOUNT = 500.00;
private static final long ONE_MINUTE = 60 * 1000;
@Override
public void processElement(
Transaction transaction,
Context context,
Collector<Alert> collector) throws Exception {
Alert alert = new Alert();
alert.setId(transaction.getAccountId());
collector.collect(alert);
}
}运行作业
yangyanping@yangyaningdeAir flink-2.2.0 % ./bin/flink run examples/streaming/frauddetection-0.1.jar
Job has been submitted with JobID 592f06bc41c0f16c1148ce28c002da6a
运行此代码将为账户 3 发出欺诈警报。你应该能在任务管理器日志中看到以下输出：
2019-08-19 14:22:06,220 INFO org.apache.flink.walkthrough.common.sink.AlertSink - Alert{id=3}
2019-08-19 14:22:11,383 INFO org.apache.flink.walkthrough.common.sink.AlertSink - Alert{id=3}
2019-08-19 14:22:16,551 INFO org.apache.flink.walkthrough.common.sink.AlertSink - Alert{id=3}
2019-08-19 14:22:21,723 INFO org.apache.flink.walkthrough.common.sink.AlertSink - Alert{id=3}
2019-08-19 14:22:26,896 INFO org.apache.flink.walkthrough.common.sink.AlertSink - Alert{id=3}更多推荐
15
0- 0
已为社区贡献11条内容
所有评论(0)