env.execute("Fraud Detection");

Scala代码：

env.execute("Fraud Detection")

欺诈检测器是作为一个KeyedProcessFunction实现的。它的方法KeyedProcessFunction#processElement被调用为每一个交易事件。第一个版本在每个事务上都会产生一个警报，有人可能会说这太保守了。

本教程的下一步将引导你用更有意义的业务逻辑来扩展欺诈检测器。

Java代码：

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

    }

}

Scala代码：

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] {

  @throws[Exception]

  def processElement(

      transaction: Transaction,

      context: KeyedProcessFunction[Long, Transaction, Alert]#Context,

      collector: Collector[Alert]): Unit = {

    val alert = new Alert

    alert.setId(transaction.getAccountId)

    collector.collect(alert)

  }

}

编写一个真实的应用程序(v1)

对于第一个版本，欺诈检测器应该为任何紧随大额交易之后进行小额交易的账户输出警报。其中，小额是指小于1.00美元的交易，大额是指超过500美元的交易。想象一下，您的欺诈检测器为一个特定的账户处理了以下交易流。

交易3和4应被标记为欺诈，因为它是一笔小额交易，0.09美元，然后是一笔大额交易，510美元。另外，交易7、8、9不是欺诈，因为0.02美元的小额交易之后并没有紧接着大额交易，而是有一个中间交易打破了这个模式。

要做到这一点，欺诈检测器必须记住跨事件的信息；只有当前一个交易金额较小时，大额交易才是欺诈的。跨事件记忆信息需要状态，这就是为什么我们决定使用KeyedProcessFunction。它提供了对状态和时间的精细控制，这将使我们在整个演练过程中以更复杂的要求来演化我们的算法。

最直接的实现是一个布尔标志，每当处理一个小的事务时，就会设置这个标志。当有大额交易通过时，你可以简单地检查该账户的标志是否被设置。

然而，仅仅将标志作为FraudDetector类中的成员变量来实现是不行的。Flink用FraudDetector的同一个对象实例来处理多个账户的交易，这意味着如果账户A和账户B通过FraudDetector的同一个实例进行路由，那么账户A的交易可能会将flag设置为true，然后账户B的交易可能会触发虚假警报。当然，我们可以使用像Map这样的数据结构来跟踪各个键的标志，但是，一个简单的成员变量并不具备容错性，一旦发生故障，它的所有信息就会丢失。因此，如果应用程序不得不重新启动以从故障中恢复，欺诈检测器可能会错过警报。

为了解决这些挑战，Flink提供了容错状态的基元，这些基元几乎和普通成员变量一样容易使用。

Flink中最基本的状态类型是ValueState，这是一种为其封装的任何变量增加容错性的数据类型。ValueState是键控状态的一种形式，这意味着它只有在键控上下文中应用的操作符中才可用；任何操作符紧跟DataStream#keyBy。一个操作符的键控状态会自动作用于当前处理的记录的键。在这个例子中，键是当前交易的账户id（由keyBy()声明），FraudDetector为每个账户维护一个独立的状态。ValueState是使用ValueStateDescriptor创建的，它包含了关于Flink应该如何管理变量的元数据。该状态应该在函数开始处理数据之前被注册。正确的钩子是open()方法。

Java代码：

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

    }

Scala代码：

@SerialVersionUID(1L)

class FraudDetector extends KeyedProcessFunction[Long, Transaction, Alert] {

  @transient private var flagState: ValueState[java.lang.Boolean] = _

  @throws[Exception]

  override def open(parameters: Configuration): Unit = {

    val flagDescriptor = new ValueStateDescriptor("flag", Types.BOOLEAN)

    flagState = getRuntimeContext.getState(flagDescriptor)

  }

ValueState是一个包装类，类似于Java标准库中的AtomicReference或AtomicLong。它提供了三个方法来与它的内容进行交互；update设置状态，value获取当前值，clear删除其内容。如果某个键的状态为空，比如在应用程序开始时或调用ValueState#clear后，那么ValueState#value将返回null。对ValueState#value返回的对象的修改不能保证被系统识别，因此所有的修改都必须使用ValueState#update进行。否则，容错是由Flink在引擎盖下自动管理的，所以你可以像与任何标准变量一样与它交互。

下面，你可以看到一个例子，说明你如何使用标志状态来跟踪潜在的欺诈交易。

Java代码：

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

    }

}

Scala代码：

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

    flagState.clear()

  }

  if (transaction.getAmount < FraudDetector.SMALL_AMOUNT) {

    // set the flag to true

    flagState.update(true)

  }

}

对于每一笔交易，欺诈检测器都会检查该账户的标志状态。请记住，ValueState总是被扩展到当前的键，即账户。如果标志是非空的，那么该账户最后一次看到的交易金额很小，所以如果这次交易的金额很大，那么检测器就会输出一个欺诈警报。

经过该检查后，标志状态无条件清零。要么是当前交易引起了欺诈警报，模式结束；要么是当前交易没有引起警报，模式被破坏，需要重新开始。

最后，检查交易金额是否偏小。如果是，则设置标志，以便下一个事件检查。注意ValueState<Boolean>实际上有三种状态，unset ( null)、true和false，因为所有ValueState的都是可空的。这个工作只利用unset ( null)和true来检查标志是否被设置。

欺诈检测器v2: 状态+时间=️

骗子不会等很久才进行大额购买，以减少他们的测试交易被发现的机会。例如，假设你想给你的欺诈检测器设置一个1分钟的超时时间；也就是说，在前面的例子中，只有当交易3和4发生在1分钟之内时，它们才会被认为是欺诈。Flink的KeyedProcessFunction允许你设置定时器，在未来的某个时间点调用回调方法。

让我们看看如何修改我们的Job以符合我们的新要求。

每当标志被设置为真时，就设置一个未来1分钟的定时器。
当定时器发射时，通过清除标志的状态来重置标志。
如果标志曾经被清除，则应取消定时器。
要取消一个定时器，你必须记住它被设置的时间，而记住意味着状态，所以你将首先创建一个定时器状态以及你的标志状态。

Java代码：

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

}

Scala代码：

@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)

  }

KeyedProcessFunction#processElement是用一个包含定时器服务的Context调用的。定时器服务可以用来查询当前时间、注册定时器和删除定时器。有了它，你可以在每次设置标志时，设置未来1分钟的定时器，并将时间戳存储在timerState中。

Java代码：

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

}

Scala代码：

if (transaction.getAmount < FraudDetector.SMALL_AMOUNT) {

  // set the flag to true

  flagState.update(true)

  // set the timer and timer state

  val timer = context.timerService.currentProcessingTime + FraudDetector.ONE_MINUTE

  context.timerService.registerProcessingTimeTimer(timer)

  timerState.update(timer)

}

Java代码：

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

}

Scala代码：

@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()

}

就是这样，一个全功能、有状态、分布式的流媒体应用!

最终程序

Java代码：

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

    }

}

Scala代码：

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

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()

  }

}

预期产出

使用所提供的TransactionSource运行这段代码将为账户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}