源码网站代理,企业自助建站哪家好,网站建设运营协议,东莞建筑公司Flink 风险欺诈检测 文章目录 Flink 风险欺诈检测背景准备条件FraudDetectionJob.javaFraudDetector.java 代码分析执行环境创建数据源对事件分区 欺诈检测输出结果运行作业欺诈检测器 欺诈检测器 v1#xff1a;状态欺诈检测器 v2#xff1a;状态 时间完整的程序期望的…Flink 风险欺诈检测 文章目录 Flink 风险欺诈检测背景准备条件FraudDetectionJob.javaFraudDetector.java 代码分析执行环境创建数据源对事件分区 欺诈检测输出结果运行作业欺诈检测器 欺诈检测器 v1状态欺诈检测器 v2状态 时间完整的程序期望的结果 Apache Flink 提供了 DataStream API 来实现稳定可靠的、有状态的流处理应用程序。 Flink 支持对状态和时间的细粒度控制以此来实现复杂的事件驱动数据处理系统。 这个入门指导手册讲述了如何通过 Flink DataStream API 来实现一个有状态流处理程序。 背景
在当今数字时代信用卡欺诈行为越来越被重视。 罪犯可以通过诈骗或者入侵安全级别较低系统来盗窃信用卡卡号。 用盗得的信用卡进行很小额度的例如一美元或者更小额度的消费进行测试。 如果测试消费成功那么他们就会用这个信用卡进行大笔消费来购买一些他们希望得到的或者可以倒卖的财物。
在这个教程中你将会建立一个针对可疑信用卡交易行为的反欺诈检测系统。 通过使用一组简单的规则你将了解到 Flink 如何为我们实现复杂业务逻辑并实时执行。
准备条件
这个代码练习假定你对 Java 有一定的了解当然如果你之前使用的是其他开发语言你也应该能够跟随本教程进行学习。
Flink提供了一个准备好的 Flink Maven Archetype 能够快速创建一个包含了必要依赖的 Flink 程序骨架基于此你可以把精力集中在编写业务逻辑上即可。 这些已包含的依赖包括 flink-streaming-java、flink-walkthrough-common 等他们分别是 Flink 应用程序的核心依赖项和这个代码练习需要的数据生成器
mvn archetype:generate \-DarchetypeGroupIdorg.apache.flink \-DarchetypeArtifactIdflink-walkthrough-datastream-java \-DarchetypeVersion1.18.0 \-DgroupIdfrauddetection \-DartifactIdfrauddetection \-Dversion0.1 \-Dpackagespendreport \-DinteractiveModefalse可以在命令上里执行执行上述命令
‘
你可以根据自己的情况修改 groupId、 artifactId 和 package。通过这三个参数 Maven 将会创建一个名为 frauddetection 的文件夹包含了所有依赖的整个工程项目将会位于该文件夹下。 将工程目录导入到你的开发环境之后你可以找到 FraudDetectionJob.java 代码文件文件中的代码如下所示。你可以在 IDE 中直接运行这个文件。 同时你可以试着在数据流中设置一些断点或者以 DEBUG 模式来运行程序体验 Flink 是如何运行的。
创建好之后可以在IDEA 中打开该项目尝试运行
在 IDE 中运行该项目可能会遇到 java.langNoClassDefFoundError 的异常。这很可能是因为运行所需要的 Flink 的依赖库没有默认被全部加载到类路径classpath里。
’
IntelliJ IDE前往 运行 编辑配置 修改选项 选中 将带有 “provided” 范围的依赖项添加到类路径。这样的话运行配置将会包含所有在 IDE 中运行所必须的类。
‘
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;public class FraudDetectionJob {public static void main(String[] args) throws Exception {StreamExecutionEnvironment env StreamExecutionEnvironment.getExecutionEnvironment();DataStreamTransaction transactions env.addSource(new TransactionSource()).name(transactions);DataStreamAlert 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;public class FraudDetector extends KeyedProcessFunctionLong, 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;Overridepublic void processElement(Transaction transaction,Context context,CollectorAlert collector) throws Exception {Alert alert new Alert();alert.setId(transaction.getAccountId());collector.collect(alert);}
}代码分析
让我们一步步地来分析一下这两个代码文件。FraudDetectionJob 类定义了程序的数据流而 FraudDetector 类定义了欺诈交易检测的业务逻辑。
下面我们开始讲解整个 Job 是如何组装到 FraudDetectionJob 类的 main 函数中的。
执行环境
第一行的 StreamExecutionEnvironment 用于设置你的执行环境。 任务执行环境用于定义任务的属性、创建数据源以及最终启动任务的执行。
StreamExecutionEnvironment env StreamExecutionEnvironment.getExecutionEnvironment();创建数据源
数据源从外部系统例如 Apache Kafka、Rabbit MQ 或者 Apache Pulsar 接收数据然后将数据送到 Flink 程序中。 这个代码练习使用的是一个能够无限循环生成信用卡模拟交易数据的数据源。 每条交易数据包括了信用卡 ID accountId交易发生的时间 timestamp 以及交易的金额amount。 绑定到数据源上的 name 属性是为了调试方便如果发生一些异常我们能够通过它快速定位问题发生在哪里。
DataStreamTransaction transactions env.addSource(new TransactionSource()).name(transactions);对事件分区 欺诈检测
transactions 这个数据流包含了大量的用户交易数据需要被划分到多个并发上进行欺诈检测处理。由于欺诈行为的发生是基于某一个账户的所以必须要保证同一个账户的所有交易行为数据要被同一个并发的 task 进行处理。
为了保证同一个 task 处理同一个 key 的所有数据你可以使用 DataStream#keyBy 对流进行分区。 process() 函数对流绑定了一个操作这个操作将会对流上的每一个消息调用所定义好的函数。 通常一个操作会紧跟着 keyBy 被调用在这个例子中这个操作是FraudDetector该操作是在一个 keyed context 上执行的。
DataStreamAlert 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 接口的一个实现。 他的方法 KeyedProcessFunction#processElement 将会在每个交易事件上被调用。 这个程序里边会对每笔交易发出警报有人可能会说这做报过于保守了。
本教程的后续步骤将指导你对这个欺诈检测器进行更有意义的业务逻辑扩展。
public class FraudDetector extends KeyedProcessFunctionLong, 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;Overridepublic void processElement(Transaction transaction,Context context,CollectorAlert collector) throws Exception {Alert alert new Alert();alert.setId(transaction.getAccountId());collector.collect(alert);}
}这里我们的逻辑非常简单其实就是直接传递给下游了也就是每个数据我们都无区别的做了告警
欺诈检测器 v1状态
我们先实现第一版报警程序对于一个账户如果出现小于 $1 美元的交易后紧跟着一个大于 $500 的交易就输出一个报警信息。
假设你的欺诈检测器所处理的交易数据如下 交易 3 和交易 4 应该被标记为欺诈行为因为交易 3 是一个 $0.09 的小额交易而紧随着的交易 4 是一个 $510 的大额交易。 另外交易 7、8 和 交易 9 就不属于欺诈交易了因为在交易 7 这个 $0.02 的小额交易之后并没有跟随一个大额交易而是一个金额适中的交易这使得交易 7 到 交易 9 不属于欺诈行为。
欺诈检测器需要在多个交易事件之间记住一些信息。仅当一个大额的交易紧随一个小额交易的情况发生时这个大额交易才被认为是欺诈交易。 在多个事件之间存储信息就需要使用到 状态这也是我们选择使用 KeyedProcessFunction 的原因。 它能够同时提供对状态和时间的细粒度操作这使得我们能够在接下来的代码练习中实现更复杂的算法。
最直接的实现方式是使用一个 boolean 型的标记状态来表示是否刚处理过一个小额交易。 当处理到该账户的一个大额交易时你只需要检查这个标记状态来确认上一个交易是是否小额交易即可。
然而仅使用一个标记作为 FraudDetector 的类成员来记录账户的上一个交易状态是不准确的。 Flink 会在同一个 FraudDetector 的并发实例中处理多个账户的交易数据假设当账户 A 和账户 B 的数据被分发的同一个并发实例上处理时账户 A 的小额交易行为可能会将标记状态设置为真随后账户 B 的大额交易可能会被误判为欺诈交易。 当然我们可以使用如 Map 这样的数据结构来保存每一个账户的状态但是常规的类成员变量是无法做到容错处理的当任务失败重启后之前的状态信息将会丢失。 这样的话如果程序曾出现过失败重启的情况将会漏掉一些欺诈报警。
为了应对这个问题Flink 提供了一套支持容错状态的原语这些原语几乎与常规成员变量一样易于使用。
Flink 中最基础的状态类型是 ValueState这是一种能够为被其封装的变量添加容错能力的类型。 ValueState 是一种 keyed state也就是说它只能被用于 keyed context 提供的 operator 中即所有能够紧随 DataStream#keyBy 之后被调用的operator。 一个 operator 中的 keyed state 的作用域默认是属于它所属的 key 的。 这个例子中key 就是当前正在处理的交易行为所属的信用卡账户key 传入 keyBy() 函数调用而 FraudDetector 维护了每个帐户的标记状态。 ValueState 需要使用 ValueStateDescriptor 来创建ValueStateDescriptor 包含了 Flink 如何管理变量的一些元数据信息。状态在使用之前需要先被注册。 状态需要使用 open() 函数来注册状态。
public class FraudDetector extends KeyedProcessFunctionLong, Transaction, Alert {private static final long serialVersionUID 1L;private transient ValueStateBoolean flagState;Overridepublic void open(Configuration parameters) {ValueStateDescriptorBoolean flagDescriptor new ValueStateDescriptor(flag,Types.BOOLEAN);flagState getRuntimeContext().getState(flagDescriptor);}ValueState 是一个包装类类似于 Java 标准库里边的 AtomicReference 和 AtomicLong。 它提供了三个用于交互的方法。update 用于更新状态value 用于获取状态值还有 clear 用于清空状态。 如果一个 key 还没有状态例如当程序刚启动或者调用过 ValueState#clear 方法时ValueState#value 将会返回 null。 如果需要更新状态需要调用 ValueState#update 方法直接更改 ValueState#value 的返回值可能不会被系统识别。 容错处理将在 Flink 后台自动管理你可以像与常规变量那样与状态变量进行交互。
下边的示例说明了如何使用标记状态来追踪可能的欺诈交易行为。
Override
public void processElement(Transaction transaction,Context context,CollectorAlert collector) throws Exception {// Get the current state for the current keyBoolean lastTransactionWasSmall flagState.value();// Check if the flag is setif (lastTransactionWasSmall ! null) {if (transaction.getAmount() LARGE_AMOUNT) {// Output an alert downstreamAlert alert new Alert();alert.setId(transaction.getAccountId());collector.collect(alert);}// Clean up our stateflagState.clear();}if (transaction.getAmount() SMALL_AMOUNT) {// Set the flag to trueflagState.update(true);}
}对于每笔交易欺诈检测器都会检查该帐户的标记状态。 请记住ValueState 的作用域始终限于当前的 key即信用卡帐户。 如果标记状态不为空则该帐户的上一笔交易是小额的因此如果当前这笔交易的金额很大那么检测程序将输出报警信息。
在检查之后不论是什么状态都需要被清空。 不管是当前交易触发了欺诈报警而造成模式的结束还是当前交易没有触发报警而造成模式的中断都需要重新开始新的模式检测。
最后检查当前交易的金额是否属于小额交易。 如果是那么需要设置标记状态以便可以在下一个事件中对其进行检查。 注意ValueStateBoolean 实际上有 3 种状态unset (null)true和 falseValueState 是允许空值的。 我们的程序只使用了 unset (null) 和 true 两种来判断标记状态被设置了与否。
欺诈检测器 v2状态 时间
骗子们在小额交易后不会等很久就进行大额消费这样可以降低小额测试交易被发现的几率。 比如假设你为欺诈检测器设置了一分钟的超时对于上边的例子交易 3 和 交易 4 只有间隔在一分钟之内才被认为是欺诈交易。 Flink 中的 KeyedProcessFunction 允许您设置计时器该计时器在将来的某个时间点执行回调函数。
让我们看看如何修改程序以符合我们的新要求
当标记状态被设置为 true 时设置一个在当前时间一分钟后触发的定时器。当定时器被触发时重置标记状态。当标记状态被重置时删除定时器。
要删除一个定时器你需要记录这个定时器的触发时间这同样需要状态来实现所以你需要在标记状态后也创建一个记录定时器时间的状态。
Java
private transient ValueStateBoolean flagState;
private transient ValueStateLong timerState;Override
public void open(Configuration parameters) {ValueStateDescriptorBoolean flagDescriptor new ValueStateDescriptor(flag,Types.BOOLEAN);flagState getRuntimeContext().getState(flagDescriptor);ValueStateDescriptorLong timerDescriptor new ValueStateDescriptor(timer-state,Types.LONG);timerState getRuntimeContext().getState(timerDescriptor);
}KeyedProcessFunction#processElement 需要使用提供了定时器服务的 Context 来调用。 定时器服务可以用于查询当前时间、注册定时器和删除定时器。 使用它你可以在标记状态被设置时也设置一个当前时间一分钟后触发的定时器同时将触发时间保存到 timerState 状态中。
if (transaction.getAmount() SMALL_AMOUNT) {// set the flag to trueflagState.update(true);// set the timer and timer statelong timer context.timerService().currentProcessingTime() ONE_MINUTE;context.timerService().registerProcessingTimeTimer(timer);timerState.update(timer);
}处理时间是本地时钟时间这是由运行任务的服务器的系统时间来决定的。
当定时器触发时将会调用 KeyedProcessFunction#onTimer 方法。 通过重写这个方法来实现一个你自己的重置状态的回调逻辑。
Override
public void onTimer(long timestamp, OnTimerContext ctx, CollectorAlert out) {// remove flag after 1 minutetimerState.clear();flagState.clear();
}最后如果要取消定时器你需要删除已经注册的定时器并同时清空保存定时器的状态。 你可以把这些逻辑封装到一个助手函数中而不是直接调用 flagState.clear()。
private void cleanUp(Context ctx) throws Exception {// delete timerLong timer timerState.value();ctx.timerService().deleteProcessingTimeTimer(timer);// clean up all statetimerState.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 KeyedProcessFunctionLong, 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 ValueStateBoolean flagState;private transient ValueStateLong timerState;Overridepublic void open(Configuration parameters) {ValueStateDescriptorBoolean flagDescriptor new ValueStateDescriptor(flag,Types.BOOLEAN);flagState getRuntimeContext().getState(flagDescriptor);ValueStateDescriptorLong timerDescriptor new ValueStateDescriptor(timer-state,Types.LONG);timerState getRuntimeContext().getState(timerDescriptor);}Overridepublic void processElement(Transaction transaction,Context context,CollectorAlert collector) throws Exception {// Get the current state for the current keyBoolean lastTransactionWasSmall flagState.value();// Check if the flag is setif (lastTransactionWasSmall ! null) {if (transaction.getAmount() LARGE_AMOUNT) {//Output an alert downstreamAlert alert new Alert();alert.setId(transaction.getAccountId());collector.collect(alert);}// Clean up our statecleanUp(context);}if (transaction.getAmount() SMALL_AMOUNT) {// set the flag to trueflagState.update(true);long timer context.timerService().currentProcessingTime() ONE_MINUTE;context.timerService().registerProcessingTimeTimer(timer);timerState.update(timer);}}Overridepublic void onTimer(long timestamp, OnTimerContext ctx, CollectorAlert out) {// remove flag after 1 minutetimerState.clear();flagState.clear();}private void cleanUp(Context ctx) throws Exception {// delete timerLong timer timerState.value();ctx.timerService().deleteProcessingTimeTimer(timer);// clean up all statetimerState.clear();flagState.clear();}
}期望的结果
使用已准备好的 TransactionSource 数据源运行这个代码将会检测到账户 3 的欺诈行为并输出报警信息。 你将能够在你的 task manager 的日志中看到下边输出
2019-08-19 14:22:06,220 INFO org.apache.flink.walkthrough.common.sink.AlertSink - Alert{id3}
2019-08-19 14:22:11,383 INFO org.apache.flink.walkthrough.common.sink.AlertSink - Alert{id3}
2019-08-19 14:22:16,551 INFO org.apache.flink.walkthrough.common.sink.AlertSink - Alert{id3}
2019-08-19 14:22:21,723 INFO org.apache.flink.walkthrough.common.sink.AlertSink - Alert{id3}
2019-08-19 14:22:26,896 INFO org.apache.flink.walkthrough.common.sink.AlertSink - Alert{id3}
