Apache Flink 提供了一个 DataStream API，用于构建健壮的、有状态的流应用程序。它提供了对状态和时间的细粒度控制，允许实现高级的 事件驱动（event-driven）系统。在这里会逐步指导你学习如何使用Flink 的 DataStream API 构建一个有状态的流式应用程序。

What Are You Building?

在数字时代，信用卡欺诈日益受到关注。犯罪分子通过诈骗或入侵不安全的系统窃取用户的信用卡号码。在盗取之后，他们通常会通过一次会多次的小额购物来对被盗的信用卡进行测试，如果确认有效的话，他们则会进行更大的购买去获得可以据为己有或出售的东西。

在本教程中，你将会构建一个欺诈检测系统 (a fraud detection system)，用于对可疑的信用卡交易进行告警。通过使用一组简单的规则，你将可以看到Flink如何允许我们实现高级业务逻辑并且实时的执行计算。

Prerequisites

本次演练假定您对Java或者Scala有一定的了解，不过即使你之前掌握的是其他不同的语言，也应该能够理解。

Help, I’m Stuck!

如果你陷入困境，可以查看 community support resources。值得一提的是，Apache Flink 的用户邮件列表一直被评为所有Apache项目中最活跃的项目之一，因此通过邮件列表进行求助吗，也不失为一种很棒的快速解决问题途径。

How to Follow Along

如果你想继续跟进本教程，那么则需要一台具备以下环境计算机：

• Java 8 或者 Java 11
• Maven

提供的Flink maven工程模板会快速创建一个包含所有依赖的项目框架，因此您只需要专注于填写业务逻辑即可。
这些项目依赖包括 flink-streaming-java,它是所有应用程序的核心依赖项，以及flink-walkthrough-common，它具有特定于此 walkthrough的数据生成器和其他类。

如果你乐意的话可以修改下面 groupId、artfactId和 package的值。如果使用下面的默认参数，maven将会创建一个名为 frauddetection 的目录，
该目录包含了一个具有完成本教程所需所有依赖的一个项目。将该项目导入编辑器后，可以找到一个名为 FraudDetectionJob.java的文件，里面的代码可以直接在你的IDE中运行。
尝试在数据流处理代码中设置断点，在DEBUG 模式下调试代码以了解一切是如何工作的。

 
 mvn archetype:generate \
    -DarchetypeGroupId=org.apache.flink \
    -DarchetypeArtifactId=flink-walkthrough-datastream-java \
    -DarchetypeVersion=1.12.0 \
    -DgroupId=frauddetection \
    -DartifactId=frauddetection \
    -Dversion=0.1 \
    -Dpackage=spendreport \
    -DinteractiveMode=false
    
    
[INFO] ----------------------------------------------------------------------------
[INFO] Using following parameters for creating project from Archetype: flink-walkthrough-datastream-java:1.12.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: version, Value: 0.1
[INFO] Parameter: groupId, Value: frauddetection
[INFO] Parameter: artifactId, Value: frauddetection
[WARNING] CP Don't override file /usr/local/flink/flink-1.12.2/frauddetection/src/main/resources
[INFO] Project created from Archetype in dir: /usr/local/flink/flink-1.12.2/frauddetection
[INFO] ------------------------------------------------------------------------
[INFO] BUILD SUCCESS
[INFO] ------------------------------------------------------------------------
[INFO] Total time:  55.333 s
[INFO] Finished at: 2021-04-19T10:48:49+08:00
[INFO] ------------------------------------------------------------------------

将项目到 IDEA

image.png

FraudDetectionJob.java

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

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);
    }
}

Breaking Down the Code

让我们逐步来浏览这两个文件的代码。FraudDetection类 定义了应用程序的数据流， FraudDetector类定义了检测欺诈事务的业务逻辑。

下面开始讲述如何在 FraudDetectionJob 类的主方法中组装作业。

执行环境

第一行的StreamExecutionEnvironment用于设置你的执行环境。 任务执行环境用于定义任务的属性、创建数据源以及最终启动任务的执行。

StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

创建数据源

数据源从外部系统例如 Apache Kafka、Rabbit MQ 或者 Apache Pulsar 接收数据，然后将数据送到 Flink 程序中。 这个代码练习使用的是一个能够无限循环生成信用卡模拟交易数据的数据源。 每条交易数据包括了信用卡 ID （accountId），交易发生的时间 （timestamp） 以及交易的金额（amount）。 绑定到数据源上的name属性是为了调试方便，如果发生一些异常，我们能够通过它快速定位问题发生在哪里。

DataStream<Transaction> transactions = env
    .addSource(new TransactionSource())
    .name("transactions");

对事件分区 & 欺诈检测

transactions数据流包含了大量用户的交易订单数据，因此需要启动多个欺诈检测的任务对其进行并行处理。
由于欺诈行为的发生是基于某一个账户的，所欲你必须保证相同账户的所有交易数据都被分配到同一个欺诈检测的并行任务当中去处理。

为了确保同一个key的所有记录被相同的物理上的 task处理，你可以使用 DataStream#keyby对数据流进行分区。
process() 函数 可以对数据流中每一个分区的元素调用预先绑定好的函数。通常来说，一个操作会紧跟着在keyby之后调用，在这个例子中，这个操作是 FraudDetector ， 它在 keyed context中执行

DataStream<Alert> alerts = transactions
    .keyBy(Transaction::getAccountId)
    .process(new FraudDetector())
    .name("fraud-detector");

输出结果

sink会将 DataStream 写到外部系统，例如 Apache Kafka、Cassandra或者AWS kinesis等。AlertSink使用**INFO **级别的日志打印每一个Alert的数据记录，而不是将其写入持久存储，以便你可以方便地查看结果。

alerts.addSink(new AlertSink());

运行作业

Flink 应用程序是惰性加载的，只有当全部构建完成之后才会发布到集群上去运行。通过调用 StreamExecutionEnvironment#execute并给他一个名称就可以开始运行我们的任务。

env.execute("Fraud Detection");

欺诈检测器

FraudDetector实现了 KeyedProcessFunction接口。它会为每个交易事件调用 processElement方法，本程序（第一个版本）会在每一个交易事件上触发一个告警，有些人可能会说这个告警过于保守了。

本教程的后续步骤将会指导你扩展这个欺诈检测程序，让它3具备更有意义的业务逻辑。

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);
    }
}

实现一个真正的应用程序（V1）

在第一个版本中，欺诈检测程序将会对任何一个 在小于1美元的交易之后紧接着发生一个大于500美元的交易的账户触发告警。
假设你的欺诈检测程序为特定账号处理以下交易数据流：

image.png

交易3和交易4会被标记为欺诈事件，因为在0.09美元的交易之后发生了510美元的大额交易。需要特别支出的是，交易7,8,9不会被认为是欺诈行为，因为在0.02这个小额交易的后面紧跟发生的是30.01美元的交易，交易9前面的也并不是小于1美元的小额交易。

由于只有在前一个小的交易之后的大额交易才会被判定位位欺诈行为，要做到这一点检测程序就必须能够记录跨事件的信息。记住跨事件的信息需要状态算子，这也是我们决定使用 keyedProcessFunction的理由。它提供了对状态和时间的细粒度控制。这就允许我们在本次演练中可以使用更复杂的需求来改进我们的算法。

最直接的实现方式是设置一个boolean型的状态标记来表明是否刚处理过一个小额交易。当一个大额交易产生时，你可以通过检查这个布尔标记来确定上一个交易是否是小额交易。

然而，仅使用一个标记作为 FraudDetector的类成员来记录账户的上一个交易状态是不准确的。Flink会在同一个FraudDetector的并发示例中处理多个账户的交易数据。假设，当账户A和账户B的数据备份发到同一个并发实例上处理时，账户A的小额交易行为将会使得boolean型状态标记为true，如果随后处理的是账户B的大额交易，则该交易会被误判为欺诈行为。当然，我们可以使用Map这样的数据结构来保存每一个账户的状态，但是常规的类成员变量无法做到容错处理，当任务失败重启后，之前的状态将会丢失。因此，如果应用程序必须重新启动才能从故障中恢复，欺诈检测程序可能会丢失部分告警。

为了应对这些挑战，Flink提供了一套具有容错状态的原语，这些原语几乎与常规成员变量一样易于使用。

Flink中最基础的状态原语是 ValueState, 这是一个可以为其包装的任何变量添加容错性的数据类型。ValueState是一种 keyed state,这意味着它只有 在 被应用在keyed上下文的算子上才可用，就是那些任何紧跟在DataStream#keyby之后调用的算子。A _keyed state _of an operator is automatically scoped to the key of the record that is currently processed. 在本例中，key 是当前事务的账户ID（as declared by keyBy()）,FraudDetector为每个账户维护一个独立的状态。ValueState需要使用 ValueStateDescriptor来创建，它包含了Flink如何管理变量的一些元数据信息。状态需要在函数开始处理数据之前调用open() 方法进行注册。

public class FraudDetector extends KeyedProcessFunction<Long, Transaction, Alert> {

    private static final long serialVersionUID = 1L;

    private transient ValueState<Boolean> flagState;

    @Override
    public void open(Configuration parameters) {
        ValueStateDescriptor<Boolean> flagDescriptor = new ValueStateDescriptor<>(
                "flag",
                Types.BOOLEAN);
        flagState = getRuntimeContext().getState(flagDescriptor);
    }
}

ValueSate 是一个包装类，类似于Java标准库里边的AtomicReference和 AtomicLong.它提供了三个可以用于交互的方法，update用于更新状态，value用于获取状态值，还有clear用于清空状态。如果一个key还没有状态，例如当程序刚启动或者调用过 ValueState#clear之后，ValueState#value将会返回 null。如果需要更新状态，需要调用ValueSate#update方法，直接更改ValueState#value的返回值可能不会被系统识别。容错处理将在Flink后台自动管理，你可以像在与常规变量那样与状态变量进行交互。

下边的示例，说明了如何使用标记状态来追踪可能的欺诈交易行为。

@Override
public void processElement(
        Transaction transaction,
        Context context,
        Collector<Alert> collector) throws Exception {

    // Get the current state for the current key
    Boolean lastTransactionWasSmall = flagState.value();

    // Check if the flag is set
    if (lastTransactionWasSmall != null) {
        if (transaction.getAmount() > LARGE_AMOUNT) {
            // Output an alert downstream
            Alert alert = new Alert();
            alert.setId(transaction.getAccountId());

            collector.collect(alert);
        }

        // Clean up our state
        flagState.clear();
    }

    if (transaction.getAmount() < SMALL_AMOUNT) {
        // Set the flag to true
        flagState.update(true);
    }
}

对于每一笔交易，欺诈检测程序都会检查该账户的标志状态。记住，ValueState对于每个账户来说作用域始终是当前的 key。如果该标志非非空值，且上一笔交易的数额很小，并且当前交易数额较大，欺诈检测程序则会触发告警。

在检查之后，不论是什么状态，都需要被清空。不管是当前交易触发了欺诈报警而造成的模式的结束还是当前交易没有从触发报警而造成的模式中断，都需要重新开始新的模式检测。

最后，价差当前交易的金额是否属于小额交易。如果是，那么需要设置标记状态，以便可以在下一个事件中对其进行检查。注意，ValueState<Boolean>实际上有三种状态: unset(null)，true 和 false， ValueState是允许空值的。我们的程序只使用了unset(null) 和 true 两种来判断标记状态被设置了与否。

欺诈检测程序（V2）状态 + 时间 =

骗子们不会等待太久在进行大额消费之前以减小他们的测试交易被发现的概率。比如，假设你为欺诈检测程序设置了1分钟的超时时间，对于上边的例子，交易3和交易4只有间隔在一分钟之内才被认为是欺诈交易。Flink中的 keyedProcessFunction允许你设置计时器，该计时器在将来的某个时间点执行回调函数。

让我们看看如何修改程序以符合我们的新要求：

• 当标记状态被设置为 true 时，设置一个在当前时间一分钟后触发的定时器。
• 当定时器被触发时，重置标记状态。
• 当标记状态被重置时，删除定时器。

要删除一个定时器，你需要记录这个定时器的触发时间，这样需要状态来实现，所以你需要在标记状态后也创建一个记录定时器时间的状态。

private transient ValueState<Boolean> flagState;
private transient ValueState<Long> timerState;

@Override
public void open(Configuration parameters) {
    ValueStateDescriptor<Boolean> flagDescriptor = new ValueStateDescriptor<>(
            "flag",
            Types.BOOLEAN);
    flagState = getRuntimeContext().getState(flagDescriptor);

    ValueStateDescriptor<Long> timerDescriptor = new ValueStateDescriptor<>(
            "timer-state",
            Types.LONG);
    timerState = getRuntimeContext().getState(timerDescriptor);
}

keyedProcessFunction#processElement需要使用提供了定时器服务的Context来调用。定时器服务可以用于查询当前时间、注册定时器和删除定时器。使用它，你可以在标记状态被设置时，也设置一个当前时间一分钟后触发 的定时器，同时，将触发时间保存在timerState状态中。

if (transaction.getAmount() < SMALL_AMOUNT) {
    // set the flag to true
    flagState.update(true);

    // set the timer and timer state
    long timer = context.timerService().currentProcessingTime() + ONE_MINUTE;
    context.timerService().registerProcessingTimeTimer(timer);
    timerState.update(timer);
}

处理时间是本地时钟时间，这是由运行任务的服务器的系统时间来决定的。
当定时器触发时，将会调用KeyedProcessFunction#onTimer方法。通过重写这个方法来实现一个你自己的重置状态的回调逻辑。

@Override
public void onTimer(long timestamp, OnTimerContext ctx, Collector<Alert> out) {
    // remove flag after 1 minute
    timerState.clear();
    flagState.clear();
}

最后，如果要取消定时器，你需要删除已经注册的定时器，并同时清空保存定时器的状态。你可以把这些；逻辑封装到一个助手函数中，而不是直接调用 flagState.clear()

private void cleanUp(Context ctx) throws Exception {
    // delete timer
    Long timer = timerState.value();
    ctx.timerService().deleteProcessingTimeTimer(timer);

    // clean up all state
    timerState.clear();
    flagState.clear();
}

这就是一个功能完备的，有状态的分布式流处理程序了。

完整的程序

package spendreport;

import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.typeinfo.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;

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;

    private transient ValueState<Boolean> flagState;
    private transient ValueState<Long> timerState;

    @Override
    public void open(Configuration parameters) {
        ValueStateDescriptor<Boolean> flagDescriptor = new ValueStateDescriptor<>(
                "flag",
                Types.BOOLEAN);
        flagState = getRuntimeContext().getState(flagDescriptor);

        ValueStateDescriptor<Long> timerDescriptor = new ValueStateDescriptor<>(
                "timer-state",
                Types.LONG);
        timerState = getRuntimeContext().getState(timerDescriptor);
    }

    @Override
    public void processElement(
            Transaction transaction,
            Context context,
            Collector<Alert> collector) throws Exception {

        // Get the current state for the current key
        Boolean lastTransactionWasSmall = flagState.value();

        // Check if the flag is set
        if (lastTransactionWasSmall != null) {
            if (transaction.getAmount() > LARGE_AMOUNT) {
                //Output an alert downstream
                Alert alert = new Alert();
                alert.setId(transaction.getAccountId());

                collector.collect(alert);
            }
            // Clean up our state
            cleanUp(context);
        }

        if (transaction.getAmount() < SMALL_AMOUNT) {
            // set the flag to true
            flagState.update(true);

            long timer = context.timerService().currentProcessingTime() + ONE_MINUTE;
            context.timerService().registerProcessingTimeTimer(timer);

            timerState.update(timer);
        }
    }

    @Override
    public void onTimer(long timestamp, OnTimerContext ctx, Collector<Alert> out) {
        // remove flag after 1 minute
        timerState.clear();
        flagState.clear();
    }

    private void cleanUp(Context ctx) throws Exception {
        // delete timer
        Long timer = timerState.value();
        ctx.timerService().deleteProcessingTimeTimer(timer);

        // clean up all state
        timerState.clear();
        flagState.clear();
    }
}

期望的结果

使用已准备好的 TransactionSource数据源运行这个代码，将会检测到账户3的欺诈行为，并输出报警信息。你将能够在你的 taskmanager 的日志中看到下边的输出：

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}

image.png