Fraud Detection with the DataStream API

Apache Flink offers a DataStream API for building robust, stateful streaming applications. It provides fine-grained control over state and time, which allows for the implementation of advanced event-driven systems. In this step-by-step guide you’ll learn how to build a stateful streaming application with Flink’s DataStream API.

What Are You Building?

Credit card fraud is a growing concern in the digital age. Criminals steal credit card numbers by running scams or hacking into insecure systems. Stolen numbers are tested by making one or more small purchases, often for a dollar or less. If that works, they then make more significant purchases to get items they can sell or keep for themselves.

In this tutorial, you will build a fraud detection system for alerting on suspicious credit card transactions. Using a simple set of rules, you will see how Flink allows us to implement advanced business logic and act in real-time.

Prerequisites

This walkthrough assumes that you have some familiarity with Java or Scala, but you should be able to follow along even if you are coming from a different programming language.

Help, I’m Stuck!

If you get stuck, check out the community support resources. In particular, Apache Flink’s user mailing list is consistently ranked as one of the most active of any Apache project and a great way to get help quickly.

How to Follow Along

If you want to follow along, you will require a computer with:

  • Java 8 or 11
  • Maven

A provided Flink Maven Archetype will create a skeleton project with all the necessary dependencies quickly, so you only need to focus on filling out the business logic. These dependencies include flink-streaming-java which is the core dependency for all Flink streaming applications and flink-walkthrough-common that has data generators and other classes specific to this walkthrough.

Java

  1. $ mvn archetype:generate \
  2.     -DarchetypeGroupId=org.apache.flink \
  3.     -DarchetypeArtifactId=flink-walkthrough-datastream-java \
  4.     -DarchetypeVersion=1.13.0 \
  5.     -DgroupId=frauddetection \
  6.     -DartifactId=frauddetection \
  7.     -Dversion=0.1 \
  8.     -Dpackage=spendreport \
  9.     -DinteractiveMode=false

Scala

  1. $ mvn archetype:generate \
  2.     -DarchetypeGroupId=org.apache.flink \
  3.     -DarchetypeArtifactId=flink-walkthrough-datastream-scala \
  4.     -DarchetypeVersion=1.13.0 \
  5.     -DgroupId=frauddetection \
  6.     -DartifactId=frauddetection \
  7.     -Dversion=0.1 \
  8.     -Dpackage=spendreport \
  9.     -DinteractiveMode=false

You can edit the groupId, artifactId and package if you like. With the above parameters, Maven will create a folder named frauddetection that contains a project with all the dependencies to complete this tutorial. After importing the project into your editor, you can find a file FraudDetectionJob.java (or FraudDetectionJob.scala) with the following code which you can run directly inside your IDE. Try setting break points through out the data stream and run the code in DEBUG mode to get a feeling for how everything works.

Java

FraudDetectionJob.java

  1. package spendreport;
  3. import org.apache.flink.streaming.api.datastream.DataStream;
  4. import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
  5. import org.apache.flink.walkthrough.common.sink.AlertSink;
  6. import org.apache.flink.walkthrough.common.entity.Alert;
  7. import org.apache.flink.walkthrough.common.entity.Transaction;
  8. import org.apache.flink.walkthrough.common.source.TransactionSource;
  10. public class FraudDetectionJob {
  12.     public static void main(String[] args) throws Exception {
  13.         StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
  15.         DataStream<Transaction> transactions = env
  16.             .addSource(new TransactionSource())
  17.             .name("transactions");
  19.         DataStream<Alert> alerts = transactions
  20.             .keyBy(Transaction::getAccountId)
  21.             .process(new FraudDetector())
  22.             .name("fraud-detector");
  24.         alerts
  25.             .addSink(new AlertSink())
  26.             .name("send-alerts");
  28.         env.execute("Fraud Detection");
  29.     }
  30. }

FraudDetector.java

  1. package spendreport;
  3. import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
  4. import org.apache.flink.util.Collector;
  5. import org.apache.flink.walkthrough.common.entity.Alert;
  6. import org.apache.flink.walkthrough.common.entity.Transaction;
  8. public class FraudDetector extends KeyedProcessFunction<Long, Transaction, Alert> {
  10.     private static final long serialVersionUID = 1L;
  12.     private static final double SMALL_AMOUNT = 1.00;
  13.     private static final double LARGE_AMOUNT = 500.00;
  14.     private static final long ONE_MINUTE = 60 * 1000;
  16.     @Override
  17.     public void processElement(
  18.             Transaction transaction,
  19.             Context context,
  20.             Collector<Alert> collector) throws Exception {
  22.         Alert alert = new Alert();
  23.         alert.setId(transaction.getAccountId());
  25.         collector.collect(alert);
  26.     }
  27. }

Scala

FraudDetectionJob.scala

  1. package spendreport
  3. import org.apache.flink.streaming.api.scala._
  4. import org.apache.flink.walkthrough.common.sink.AlertSink
  5. import org.apache.flink.walkthrough.common.entity.Alert
  6. import org.apache.flink.walkthrough.common.entity.Transaction
  7. import org.apache.flink.walkthrough.common.source.TransactionSource
  9. object FraudDetectionJob {
  11.   @throws[Exception]
  12.   def main(args: Array[String]): Unit = {
  13.     val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
  15.     val transactions: DataStream[Transaction] = env
  16.       .addSource(new TransactionSource)
  17.       .name("transactions")
  19.     val alerts: DataStream[Alert] = transactions
  20.       .keyBy(transaction => transaction.getAccountId)
  21.       .process(new FraudDetector)
  22.       .name("fraud-detector")
  24.     alerts
  25.       .addSink(new AlertSink)
  26.       .name("send-alerts")
  28.     env.execute("Fraud Detection")
  29.   }
  30. }

FraudDetector.scala

  1. package spendreport
  3. import org.apache.flink.streaming.api.functions.KeyedProcessFunction
  4. import org.apache.flink.util.Collector
  5. import org.apache.flink.walkthrough.common.entity.Alert
  6. import org.apache.flink.walkthrough.common.entity.Transaction
  8. object FraudDetector {
  9.   val SMALL_AMOUNT: Double = 1.00
  10.   val LARGE_AMOUNT: Double = 500.00
  11.   val ONE_MINUTE: Long     = 60 * 1000L
  12. }
  14. @SerialVersionUID(1L)
  15. class FraudDetector extends KeyedProcessFunction[Long, Transaction, Alert] {
  17.   @throws[Exception]
  18.   def processElement(
  19.       transaction: Transaction,
  20.       context: KeyedProcessFunction[Long, Transaction, Alert]#Context,
  21.       collector: Collector[Alert]): Unit = {
  23.     val alert = new Alert
  24.     alert.setId(transaction.getAccountId)
  26.     collector.collect(alert)
  27.   }
  28. }

Breaking Down the Code

Let’s walk step-by-step through the code of these two files. The FraudDetectionJob class defines the data flow of the application and the FraudDetector class defines the business logic of the function that detects fraudulent transactions.

We start describing how the Job is assembled in the main method of the FraudDetectionJob class.

The Execution Environment

The first line sets up your StreamExecutionEnvironment. The execution environment is how you set properties for your Job, create your sources, and finally trigger the execution of the Job.

Java

  1. StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

Scala

  1. val env = StreamExecutionEnvironment.getExecutionEnvironment

Creating a Source

Sources ingest data from external systems, such as Apache Kafka, Rabbit MQ, or Apache Pulsar, into Flink Jobs. This walkthrough uses a source that generates an infinite stream of credit card transactions for you to process. Each transaction contains an account ID (accountId), timestamp (timestamp) of when the transaction occurred, and US$ amount (amount). The name attached to the source is just for debugging purposes, so if something goes wrong, we will know where the error originated.

Java

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

Scala

  1. val transactions: DataStream[Transaction] = env
  2.   .addSource(new TransactionSource)
  3.   .name("transactions")

Partitioning Events & Detecting Fraud

The transactions stream contains a lot of transactions from a large number of users, such that it needs to be processed in parallel by multiple fraud detection tasks. Since fraud occurs on a per-account basis, you must ensure that all transactions for the same account are processed by the same parallel task of the fraud detector operator.

To ensure that the same physical task processes all records for a particular key, you can partition a stream using DataStream#keyBy. The process() call adds an operator that applies a function to each partitioned element in the stream. It is common to say the operator immediately after a keyBy, in this case FraudDetector, is executed within a keyed context.

Java

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

Scala

  1. val alerts: DataStream[Alert] = transactions
  2.   .keyBy(transaction => transaction.getAccountId)
  3.   .process(new FraudDetector)
  4.   .name("fraud-detector")

Outputting Results

A sink writes a DataStream to an external system; such as Apache Kafka, Cassandra, and AWS Kinesis. The AlertSink logs each Alert record with log level INFO, instead of writing it to persistent storage, so you can easily see your results.

Java

  1. alerts.addSink(new AlertSink());

Scala

  1. alerts.addSink(new AlertSink)

The Fraud Detector

The fraud detector is implemented as a KeyedProcessFunction. Its method KeyedProcessFunction#processElement is called for every transaction event. This first version produces an alert on every transaction, which some may say is overly conservative.

The next steps of this tutorial will guide you to expand the fraud detector with more meaningful business logic.

Java

  1. public class FraudDetector extends KeyedProcessFunction<Long, Transaction, Alert> {
  3.     private static final double SMALL_AMOUNT = 1.00;
  4.     private static final double LARGE_AMOUNT = 500.00;
  5.     private static final long ONE_MINUTE = 60 * 1000;
  7.     @Override
  8.     public void processElement(
  9.             Transaction transaction,
  10.             Context context,
  11.             Collector<Alert> collector) throws Exception {
  13.         Alert alert = new Alert();
  14.         alert.setId(transaction.getAccountId());
  16.         collector.collect(alert);
  17.     }
  18. }

Scala

  1. object FraudDetector {
  2.   val SMALL_AMOUNT: Double = 1.00
  3.   val LARGE_AMOUNT: Double = 500.00
  4.   val ONE_MINUTE: Long     = 60 * 1000L
  5. }
  7. @SerialVersionUID(1L)
  8. class FraudDetector extends KeyedProcessFunction[Long, Transaction, Alert] {
  10.   @throws[Exception]
  11.   def processElement(
  12.       transaction: Transaction,
  13.       context: KeyedProcessFunction[Long, Transaction, Alert]#Context,
  14.       collector: Collector[Alert]): Unit = {
  16.     val alert = new Alert
  17.     alert.setId(transaction.getAccountId)
  19.     collector.collect(alert)
  20.   }
  21. }

Writing a Real Application (v1)

For the first version, the fraud detector should output an alert for any account that makes a small transaction immediately followed by a large one. Where small is anything less than $1.00 and large is more than $500. Imagine your fraud detector processes the following stream of transactions for a particular account.

Expected Output

Running this code with the provided TransactionSource will emit fraud alerts for account 3. You should see the following output in your task manager logs:

  1. 2019-08-19 14:22:06,220 INFO  org.apache.flink.walkthrough.common.sink.AlertSink - Alert{id=3}
  2. 2019-08-19 14:22:11,383 INFO  org.apache.flink.walkthrough.common.sink.AlertSink - Alert{id=3}
  3. 2019-08-19 14:22:16,551 INFO  org.apache.flink.walkthrough.common.sink.AlertSink - Alert{id=3}
  4. 2019-08-19 14:22:21,723 INFO  org.apache.flink.walkthrough.common.sink.AlertSink - Alert{id=3}
  5. 2019-08-19 14:22:26,896 INFO  org.apache.flink.walkthrough.common.sink.AlertSink - Alert{id=3}