Detecting Unusual Credit Card Activity

Directions

1. Register the existing transactions topic as a KSQL stream:

CREATE STREAM TRANSACTIONS_RAW (ACCOUNT_ID VARCHAR, 
                                TIMESTAMP VARCHAR, 
                                CARD_TYPE VARCHAR, 
                                AMOUNT DOUBLE, 
                                IP_ADDRESS VARCHAR, 
                                TRANSACTION_ID VARCHAR) 
                           WITH (KAFKA_TOPIC='transactions',
                                 VALUE_FORMAT='JSON');

2. Repartition the stream on account_id, and use Avro for the target stream (this is optional):

CREATE STREAM TRANSACTIONS_SOURCE 
    WITH (VALUE_FORMAT='AVRO') AS 
          SELECT * 
            FROM TRANSACTIONS_RAW 
    PARTITION BY ACCOUNT_ID;

3. Register the existing stream of customer data from Oracle in the topic customers as a KSQL stream:

CREATE STREAM CUST_RAW_STREAM (ID BIGINT, 
                               FIRST_NAME VARCHAR, 
                               LAST_NAME VARCHAR, 
                               EMAIL VARCHAR, 
                               AVG_CREDIT_SPEND DOUBLE) 
              WITH (KAFKA_TOPIC='customers', 
                    VALUE_FORMAT='JSON');

4. Repartition the customer data stream by account_id to prepare for the join, and use Avro for the target stream (this is optional):

CREATE STREAM CUSTOMER_REKEYED 
    WITH (VALUE_FORMAT='AVRO') AS 
            SELECT * 
              FROM CUST_RAW_STREAM 
      PARTITION BY ID;

5. Register the partitioned customer data topic as a KSQL table used for the join with the incoming stream of transactions:

CREATE TABLE customer 
WITH (KAFKA_TOPIC='CUSTOMER_REKEYED', 
      VALUE_FORMAT='AVRO', 
      KEY='ID');

6. Join the transactions to customer information:

CREATE STREAM TRANSACTIONS_ENRICHED AS 
SELECT   T.ACCOUNT_ID, T.CARD_TYPE, T.AMOUNT, 
          C.FIRST_NAME + ' ' + C.LAST_NAME AS FULL_NAME, 
          C.AVG_CREDIT_SPEND 
FROM     TRANSACTIONS_SOURCE T 
          INNER JOIN CUSTOMER C 
          ON T.ACCOUNT_ID = C.ID;

7. Aggregate the stream of transactions for each account ID using a two-hour tumbling window, and filter for accounts in which the total spend in a two-hour period is greater than the customer’s average:

CREATE TABLE POSSIBLE_STOLEN_CARD AS 
SELECT   TIMESTAMPTOSTRING(WindowStart(), 'yyyy-MM-dd HH:mm:ss Z') AS WINDOW_START, 
           T.ACCOUNT_ID, T.CARD_TYPE, SUM(T.AMOUNT) AS TOTAL_CREDIT_SPEND, 
           T.FULL_NAME, MAX(T.AVG_CREDIT_SPEND) AS AVG_CREDIT_SPEND 
  FROM     TRANSACTIONS_ENRICHED T 
           WINDOW TUMBLING (SIZE 2 HOURS) 
  GROUP BY T.ACCOUNT_ID, T.CARD_TYPE, T.FULL_NAME 
  HAVING   SUM(T.AMOUNT) > MAX(T.AVG_CREDIT_SPEND) ;

Examine the output:

ksql> SELECT WINDOW_START, ACCOUNT_ID, CARD_TYPE, 
      TOTAL_CREDIT_SPEND, FULL_NAME, AVG_CREDIT_SPEND 
      FROM POSSIBLE_STOLEN_CARD;
2019-01-11 16:00:00 +0000 | 100019 | jcb | 90.69 | Horatius Keefe | 60.58
2019-01-11 16:00:00 +0000 | 100012 | mastercard | 84.04 | Juditha Shwalbe | 53.94
2019-01-11 16:00:00 +0000 | 100016 | maestro | 76.01 | Milo Drewes | 68.33
2019-01-11 16:00:00 +0000 | 100035 | visa-electron | 69.61 | Roxine Furminger | 59.68
…