A real time streaming implementation of markov chain based fraud detection

Fraud is a fact of life for the financial industry. Paypal did not become one of the only dotcom survivors by remaining a pure supplier of transaction engines. While final fraud determination is still in the hands of human experts, there has been much interest in automated processes that can syphon out suspicious activities for further scrutiny.  Given the level of global credit card transactions, such a problem falls squarely in the domain of big data. More than just handling the data volume, financial institutes also faces a technical challenge in being able to catch fraudulent transactions as they happen. All this points to the need for a real time streaming analysis capability.

Open source big data technologies have been advancing leaps and bounds recently, with the most recent push towards in-memory and streaming computation. This project creates a prototype implementation of a Markov chain based sequence mining algorithm, which combines in-coming credit card transactions with most recent transaction history and makes real time prediction based on the likelihood of the buying pattern. It is inspired by two excellent prior open source projects 1) Implementing a real-time data pipeline with Spark Streaming, which demonstrated the seamless coupling between Spark and Kafka for developing streaming applications, and 2) Real Time Fraud Detection with Sequence Mining, which presented a novel method for outlier detection which is in turn based on Markov Chains, Classifiers and Intrusion Detection.   No attempt has been made to improve on the author’s original algorithm. This project is focused solely on an alternative system design, which leverages newer technological options while completing a realistic streaming environment beyond a batch simulation.

The code in the project can be found at

https://github.com/bhomass/marseille

System overview

The scenario

A person’s credit card charge events are bound to spread sporadically over time. In order to capture someone’s spending incidents for the purpose of outlier detection, individual events need to be collected as they occur and prepended with historical spending incidents. This necessitates the use of both a real time stream capture device and an archive of previous events. Furthermore, a system capable of handling very large data throughput and low latency requires the fastest streaming computing device possible. The author’s choice for computing this prediction algorithm is Apache Spark, which although only recently appeared on the big data scene, is already making this fast moving industry taking an enthusiastic notice.   Spark couples very well with Kafka for data streaming, and with Hbase for persistence. Hbase is a well, established NoSql database known for its high capacity and high io throughput.

The prediction pipeline

The CCStreamingFeed class generates simulated credit card charge events and sends them into the kafka topic “real-time-cc”. Our spark code “MarkovPredictor” picks up the streaming feed, identify the customer for each event, find matching historical events and assemble the events to create a sequence of the right size (which is configured in the “app_config.properties” file. The default size is 5). The assembled sequences are piped into the Kafka topic “user-trans-history” for monitoring purposes.

The prediction process matches the input sequence against the transition matrix in the trained model, which is stored in the Hdfs file “/data/streaming_analysis/markovmodel.txt”.  Any sequence deemed to be suspicious is piped into the third Kafka topic “candidate-fraud-trans”. For demo purposes, this can be displayed in a Kafka console consumer. In a real system, this would be channeled into a down stream device, which would raise the attention of a human fraud investigator.

The training step

During training, we enter a large number of customers into the data generator (say 500). The trainer spark code “MarkovModelTrainer” picks up the large data stream, sort the data by customer id and generate 500 set of credit card spending sequences. Spark code excels in grouping and aggregation. After the total number of occurrences for each stat-to-state transition has been tallied and normalized, these numbers are stored in Hdfs as the prediction model.

Each of the system components are introduced next, followed by rehashing of the algorithms by Pranab.

Implementation overview

In this implementation, the stack includes four major components: spark, hdfs, hbase and Kafka, all clusterizable open source tools. Another words, they are built for redundancy and scalability. We start with a brief introduction to the components.  Further in depth discussions on them can be easily discovered on the web.

Spark

Spark has its academic origin in the Berkeley AMP Lab and is charting new territory in Apache history with its unprecedented high commit rate. Any industry watcher has to be well aware of its rising potential as the juggernaut of big data platform. It incorporates the omnipotent DAG pipeline structure for generalized data processing, leverages in-memory data caching for speed, relies on processing lineage for resiliency against node failure, and opens up development options to java, scala, python, as well as R programmers. It recently released its yarn implementation to maximize cluster resource usage. A concurrent project, MLlib, is incorporated into spark releases, which aims to boost both ease of use and scalability of machine learning algorithms, and make them accessible to those of us who are not PhD statisticians.

Spark itself is implemented in scala. The scala collection and function-programming paradigm is a perfect match for spark’s concept of RDD (Resilient Distributed Dataset), which applies successive grouping and reduction steps to a large data set on a cluster of high memory machines. Spark has build in capability to detect locality of hdfs files. A typical spark application interacts with a variety of data sources including hdfs, hbase, cassandra, and kafka, etc.

Once Spark has been started, the user can monitor job submissions from the web page.

One particularly interesting feature of spark is its capability for mixing in coming stream with archived data. Spark streaming uses a micro-batching technique in which it converts data collected over a pre-defined time window into a RDD to feed into the processing pipeline. The developer can dictate how rapidly to slide such a window forward in time for continuous data collection and processing.

However, it should be noted that the way streaming is done for this project is not based on a timed window, but rather on the sequence size. While a timed window works wonderfully for a continuous stream, credit card spending is more sporadic in nature. What this means at a programming level is that we would be calling RDD.reduceByKey rather than RDD.reduceByKeyAndWindow

One of the easiest ways to feed streaming data is to leverage the Kafka api. This brings the discussion to our second major component, Kafka, which is in so many ways a marvel in its own
right.

Kafak

Kafka is a queue, a publish-subscribe middleware, a streaming source, a centralized logging framework, and much more. It comes with build-in partitioning and replication over clustered hardware, command line interface and a multi-language api. In addition of partitioning, Kafka achieves its high performance by leveraging the high sequential write speed of modern disk drives (600MB.sec), OS pagecache (no jvm garbage collection), sendfile utility, use of simple read and appends to files (O (1) latency), network packet batching, and compression of batched messages.

The messages sent into Kafka are persisted for configurable period then discarded. Parallelism through partitioning gives Kafka an effectively constant performance with respect to data size.

Each message is tracked by a sequential id called offset, which is kept on a per consumer basis. Consumers can be labeled into a group. Within each group, only one consumer gets to consume a particular message. However, multiple unlabeled consumers will each receive one copy of each message independently. This consumer group concept therefore allows Kafka to fulfill the two traditional models of queuing (all consumer in one group) and publish-subscribe (no group affiliation) and any combination in between.

The producer on the other hand works with a cluster by picking which partition to write to, either in a round robin fashion, or based on some semantic rule.

Kafka has been applied as a high throughput replacement for messaging brokers (RabbitMQ), web activity tracking, metric or log aggregation, event sourcing, or a commit log. For the purpose of this article, the key advantage is its close integration with spark in the form of the KafkaUtils Class. More on this later.

Hbase

Hbase is a column family database, which most commonly uses hdfs as its persistence layer. It is linear scalable in performance and storage capacity making it an obvious candidate in big data use cases. The data set is partitioned across multiple region servers based on row keys.

Hbase and NoSql database in general cannot be understood in terms of the usual SQL database abstractions. Its high read / write efficiency is a direct result of leveraging contiguous file access (vis-à-vis kafka).  Whether designing the schema or using hbase during operations, it helps to visualize how the data exists as files.

The hbase data cell has a simple format known as keyvalue, shown below.

row key:column family: column qualifier:timestamp:value

Data is stored in the file cell by cell, ordered by the byte value of each segment. During a query, the user can specify the selection criterion using any of the keyvalue segments. By using more up front segments, the query enjoys a direct performance gain. Therefore, queries based on values are the least efficient.

This diagram illustrates that each column family is stored in its own file to maximize read and write speed of a single column family. It also illustrates the technique of shifting cell values up in position for faster query response.

The most efficient way to retrieve hbase data is though a scan, in which start and stop rows are specified. The pre-existing ordering of the row key means a scan is simply translates to an efficient contiguous file read.  The freedom to design the row key as arbitrary byte data gives the designer tremendous leeway for creative designs. Composite key design involving lumping together different application concerns into the row key makes possible the use of partial key scan. The user can narrow in on a data set by specifying a range of any one part of the composite key.

Hdfs

Hdfs is a file system built to run hadoop map reduce jobs. It redundantly stores large block (typically gigabytes) file content across hadoop data nodes according to a replication factor. A name node stores the directory meta data for the files.  A secondary name node takes snap shots of the file directory in time to allow quicker recovery in the event of a primary name node crash.

A large cluster houses the hadoop nodes on multiple racks. A good block placement policy takes into account the difference in bandwidth for intra and inter-rack traffic, and provides a tradeoff between minimizing the write cost, and maximizing data reliability, availability and aggregate read bandwidth.

The hadoop job tracker takes advantage of data locality by assigning task to run on data nodes which host the data required to each task.  The same data awareness is built into other computing platform such as spark.

Rehash of the sequence mining algorithm

The author characterizes each credit card transaction by its three prominent features:

1.    Amount spent: Low, Normal and High

2.   High vs normal price ticket item: Normal and High

3.   Time elapsed since the last transaction: Large, Normal and Small

This feature vector combined with the customer id and transaction id yields a stream of in coming data in the form of

G5XZW85212,JXCT97WT1AKR,LNL

2Q15KA83F3,TJE3MI8YIDA5,NNN

3H4E0847GL,7AK95UVB1MQ4,NHS

D20ID059R4,1Q8NWNUG33IM,HNL

The feature vector is of dimension 18 (3 x 2 x 3). You can therefore construct a 18 x 18 state matrix and collect statistics on the likelihood of any state transition between two successive credit card transactions.  The training step of the model involves collecting a relatively large number of user transactions and performing a simple aggregation on each state transition.  Once a trained model is prepared, each new transaction sequence can be judged to be either within the pattern or anomalous based on how well it fits the model over all.

In the prediction stage, the system operators can define the size of the transaction sequence to be evaluated and choose a threshold for the determination of outliers. For example, for a window size of 5 transactions, a threshold may be set such that at least 3 highly unlikely transitions is required to trigger the detection.

Code Review

There are a number of interesting code patterns which deserves special mentioning. First notice that this project is a mix of scala and java code. Any time SparkContext is involved, the author opted to code in scala, which is the native language for spark. Scala code embeds Jdk classes seamlessly, so there are no particular concerns arising from this mix.

RDD

Spark works its magic using its RDD (Resilient Distributed Dataset).  The standard spark process starts by converting ordinary data into an RDD. An example can be found in TrainingApp.

val dataRdd = sparkContext.parallelize(dataFeed, 2)

From this point, all the Spark methods can be applied one step at a time down the processing pipeline.

  val xactionByCustomer = dataRdd.map {

    line =>

      {

        val parts = line.split(",")

        (parts(0), parts(2)) // drop transaction id

      }

  }.reduceByKey((previous, current) => previous + "," + current)

  .map {

    tuple =>

      val line = tuple._1 + "," + tuple._2

      line

  }

In this pipeline, the input RDD is first split into three parts, followed by discarding of the unused second part (transaction id). Next, a reduction by key step concatenates all the transaction tokens into one long sequence, delimited by commas.

Streaming  RDD

In case of Streaming, it is called DStream. As mentioned before, Spark has a very convenient interface to kafka, which converts kafka data into a DStream. It is called KafkaUtils.

val messages = KafkaUtils.createStream[String, SingleTransaction, StringDecoder, SingleTransactionDecoder](streamingContext, kafkaParams, topics, StorageLevel.MEMORY_ONLY)

The first argument, streamingContext, contains all crucial parameters needed to define a streaming RDD, including the micro-batching window size (in seconds) and sliding rate.

The incoming data feed is automatically deserialized into a custom class called SingleTransaction. In the following statement, the spark code creates a (key, tokens) tuple, from which, additional reduction steps assembles the per customer transaction sequences, which can be scored by the chosen scorer.

val xactionByCustomer = messages.map(_._2).map {

    transaction =>

      val key = transaction.customerId

      var tokens = transaction.tokens

      (key, tokens)

  }

RDD Aggregation

During the model training process, we want to add up the occurrence of every state-to-state transition. This can be done by a four step process.

   val transitionMatrix = sequences.map {

      line =>

        val userId = line._1

        val records = line._2

        // each line goes into a loop

        // every pair transforms into key, value pair

        // transition key = tuple (sourceIndex, destinationIndex)

        // value = 1

        var stateCount = ArrayBuffer[String]()

        val numberRecords = records.length

        for (i <- 0 to numberRecords - 2) {

          val statesV = statesVar.value;

          val srcIndex = statesV.indexOf(records(i))

          val destIndex = statesV.indexOf(records(i + 1))

          val key = srcIndex + "-" + destIndex

          stateCount += key

        }

        stateCount.toList

    }

      .flatMap(c => c)

      .map(c => (c, 1))

      // reduce by key, add values

      .reduceByKey(_ + _)

In step one, the sequence of tokens is separated from the customer id. Each pair of consecutive tokens is lumped into a tuple to be used as a key later. The RDD coming out of the first step is a list of these tuples. The flatmap step breaks each list into single tuples, to be followed by the next step, which appends the number 1 behind the tuple. Lastly, the reduceByKey step adds all the 1’s for each tuple, resulting in the cumulative count of each tuple. This of course then represents the transition probability between any two states.

Hbase interface

When it comes to interfacing Spark to Hbase, we have a concern about the economy of database connections. In a spark job, a large dataset can potentially be composed of millions of individual tuples. If each tuple triggers a new Hbase connection, the connection opening step alone would overwhelm the system. Here we can take the advantage of the RDD.foreachParition() method. This way, the number of connections is reduced to just the number of work nodes in the spark cluster.

      tuple.foreachPartition {

        iterator =>

          // read prev windowSize -1 transactions from hbase and prepend to new token list

          val conf = HBaseConfiguration.create();

          val table = new HTable(conf, "cchistory");

          iterator.foreach {

            tuple =>

              val key = tuple._1

              var tokens = tuple._2

              val get = new Get(Bytes.toBytes(key));

              get.addFamily(Bytes.toBytes("transactions"));

              val result = table.get(get);

Once the Hbase connection is opened, the code goes through each transaction sequence, pick up the customer id, and look for any historical sequence for the same customer in the cchistory table.

Scorer subclassing

There are three scorers proposed by Pranab. Here we implement them as subclasses of the same trait “Scorer”. The factory pattern is implemented using the apply() method of the scala object by the same name.  The apply() method is automatically called during the constructor call in the OutlierDetector class.

val scorer = Scorer(algorithm, stateTranstionProb, states, stateSeqWindowSize)

The algorithm string in the variable s serves as the discriminator.

  def apply(s: String, stateTranstionProb: Array[Array[Double]], states: Array[String], stateSeqWindowSize: Int): Scorer = {

    if (s == "MissProbability") {

      return new MissProbability(stateTranstionProb, states, stateSeqWindowSize)

    } else if (s == "MissRate") {

      return new MissRate(stateTranstionProb, states, stateSeqWindowSize)

    } else {

      return new entropyReduction(stateTranstionProb, states, stateSeqWindowSize)

    }

  }

Summary

There is a healthy collection of clusterizable tools in the big data eco system. At the center of the collection is a high performance, programmable platform that allows a competent coder to create custom solutions, for which distributed execution is bundled in for free.  Spark is an in-memory, general-purpose data processing pipeline, which offers multi-language API, and ready integration with many supporting infrastructure layer tools.

This open source project demonstrates how a novel fraud detection algorithm involving real time streaming data can be implemented using these tools.

About Bruce Ho 

Bruce Ho is a big data enthusiast with a special interest in in-memory and streaming computing. He devoted the past year to implementation of predictive modeling and machine learning using Spark and NoSql technology. He is currently creating a data pipeline, which delivers real time user profile analysis for optimization of Ad campaigns. Look out for more of Bruce’s up coming open source projects. He was formerly a MIT/Caltech trained physicist, who later picked up java architecture, and big data.  Bruce worked in the software industry for over 10 years in various companies including Life Technologies, TeraData, and Amazon.