Capstone: Production-Grade Real-Time Fraud Detection System

Project Duration: 12-15 hours Difficulty: Advanced Technologies: Flink, Kafka, CEP, ML, RocksDB, Kubernetes, Prometheus Deliverable: Complete production-ready fraud detection pipeline

Project Overview

What You’ll Build

A real-time fraud detection system that processes millions of transactions per day and detects fraudulent activity using:

Multiple Detection Strategies: Velocity, location, amount spikes, behavioral patterns
Complex Event Processing: Pattern matching across transaction sequences
Machine Learning: Real-time feature extraction and model scoring
State Management: Petabyte-scale state with RocksDB
Production Deployment: Kubernetes, HA, monitoring, alerting

System Architecture

┌───────────────────────────────────────────────────────────────┐
│                 Fraud Detection Pipeline                      │
├───────────────────────────────────────────────────────────────┤
│                                                               │
│  ┌─────────┐      ┌──────────────────┐      ┌────────────┐  │
│  │  Kafka  │─────►│  Flink Pipeline  │─────►│   Alerts   │  │
│  │ (Source)│      │                  │      │  (Kafka)   │  │
│  └─────────┘      │  1. Enrichment   │      └────────────┘  │
│                   │  2. Feature Eng  │              │        │
│  ┌─────────┐      │  3. CEP Patterns │      ┌────────────┐  │
│  │  MySQL  │◄────►│  4. ML Scoring   │─────►│   MySQL    │  │
│  │(Lookup) │      │  5. Alerting     │      │  (Alerts)  │  │
│  └─────────┘      └──────────────────┘      └────────────┘  │
│                             │                                │
│                   ┌─────────▼─────────┐                      │
│                   │  State (RocksDB)  │                      │
│                   │  - User Profiles  │                      │
│                   │  - Tx History     │                      │
│                   │  - ML Features    │                      │
│                   └───────────────────┘                      │
│                                                               │
│  ┌──────────────────────────────────────────────────────┐   │
│  │  Monitoring: Prometheus + Grafana                    │   │
│  │  - Throughput, latency, state size                   │   │
│  │  - Checkpoint metrics, backpressure                  │   │
│  │  - Alert SLAs, fraud detection rate                  │   │
│  └──────────────────────────────────────────────────────┘   │
└───────────────────────────────────────────────────────────────┘

Part 1: Project Setup

Prerequisites

# Required tools
- Docker & Docker Compose
- Kubernetes (minikube or cloud)
- kubectl
- Maven 3.8+
- Java 11+
- Python 3.9+ (for ML model serving)

Project Structure

fraud-detection-flink/
├── pom.xml
├── src/
│   ├── main/
│   │   ├── java/
│   │   │   └── com/frauddetection/
│   │   │       ├── FraudDetectionJob.java
│   │   │       ├── models/
│   │   │       │   ├── Transaction.java
│   │   │       │   ├── FraudAlert.java
│   │   │       │   └── UserProfile.java
│   │   │       ├── sources/
│   │   │       │   └── TransactionGenerator.java
│   │   │       ├── enrichment/
│   │   │       │   └── UserProfileEnrichment.java
│   │   │       ├── features/
│   │   │       │   └── FeatureExtractor.java
│   │   │       ├── cep/
│   │   │       │   ├── VelocityFraudDetector.java
│   │   │       │   ├── LocationAnomalyDetector.java
│   │   │       │   └── DecliningPatternDetector.java
│   │   │       ├── ml/
│   │   │       │   └── MLScoringFunction.java
│   │   │       └── sinks/
│   │   │           └── AlertSink.java
│   │   └── resources/
│   │       └── flink-conf.yaml
│   └── test/
│       └── java/
│           └── com/frauddetection/
│               └── FraudDetectionTest.java
├── k8s/
│   ├── flink-cluster.yaml
│   ├── kafka.yaml
│   └── monitoring.yaml
├── docker-compose.yml
└── README.md

Maven Dependencies (pom.xml)

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0">
    <modelVersion>4.0.0</modelVersion>

    <groupId>com.frauddetection</groupId>
    <artifactId>fraud-detection-flink</artifactId>
    <version>1.0-SNAPSHOT</version>

    <properties>
        <flink.version>1.18.0</flink.version>
        <scala.binary.version>2.12</scala.binary.version>
        <java.version>11</java.version>
        <kafka.version>3.4.0</kafka.version>
    </properties>

    <dependencies>
        <!-- Flink Core -->
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-streaming-java</artifactId>
            <version>${flink.version}</version>
            <scope>provided</scope>
        </dependency>

        <!-- Flink Kafka Connector -->
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-connector-kafka</artifactId>
            <version>${flink.version}</version>
        </dependency>

        <!-- Flink CEP -->
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-cep</artifactId>
            <version>${flink.version}</version>
        </dependency>

        <!-- Flink JDBC -->
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-connector-jdbc</artifactId>
            <version>3.1.0-1.17</version>
        </dependency>

        <!-- RocksDB State Backend -->
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-statebackend-rocksdb</artifactId>
            <version>${flink.version}</version>
        </dependency>

        <!-- MySQL Driver -->
        <dependency>
            <groupId>mysql</groupId>
            <artifactId>mysql-connector-java</artifactId>
            <version>8.0.33</version>
        </dependency>

        <!-- JSON -->
        <dependency>
            <groupId>com.fasterxml.jackson.core</groupId>
            <artifactId>jackson-databind</artifactId>
            <version>2.15.2</version>
        </dependency>

        <!-- Lombok -->
        <dependency>
            <groupId>org.projectlombok</groupId>
            <artifactId>lombok</artifactId>
            <version>1.18.28</version>
            <scope>provided</scope>
        </dependency>

        <!-- Testing -->
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-test-utils</artifactId>
            <version>${flink.version}</version>
            <scope>test</scope>
        </dependency>
        <dependency>
            <groupId>junit</groupId>
            <artifactId>junit</artifactId>
            <version>4.13.2</version>
            <scope>test</scope>
        </dependency>
    </dependencies>

    <build>
        <plugins>
            <plugin>
                <groupId>org.apache.maven.plugins</groupId>
                <artifactId>maven-compiler-plugin</artifactId>
                <version>3.11.0</version>
                <configuration>
                    <source>${java.version}</source>
                    <target>${java.version}</target>
                </configuration>
            </plugin>
            <plugin>
                <groupId>org.apache.maven.plugins</groupId>
                <artifactId>maven-shade-plugin</artifactId>
                <version>3.5.0</version>
                <executions>
                    <execution>
                        <phase>package</phase>
                        <goals>
                            <goal>shade</goal>
                        </goals>
                        <configuration>
                            <transformers>
                                <transformer implementation="org.apache.maven.plugins.shade.resource.ManifestResourceTransformer">
                                    <mainClass>com.frauddetection.FraudDetectionJob</mainClass>
                                </transformer>
                            </transformers>
                        </configuration>
                    </execution>
                </executions>
            </plugin>
        </plugins>
    </build>
</project>

Part 2: Data Models

Transaction Model

package com.frauddetection.models;

import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;
import java.io.Serializable;

@Data
@NoArgsConstructor
@AllArgsConstructor
public class Transaction implements Serializable {
    private String transactionId;
    private String cardId;
    private String userId;
    private double amount;
    private String merchantId;
    private String merchantCategory;
    private String country;
    private String city;
    private double latitude;
    private double longitude;
    private String status; // APPROVED, DECLINED
    private long timestamp;
    private String deviceId;
    private String ipAddress;

    // For JSON serialization
    public static class Schema extends org.apache.flink.api.common.serialization.DeserializationSchema<Transaction>,
            org.apache.flink.api.common.serialization.SerializationSchema<Transaction> {

        private final com.fasterxml.jackson.databind.ObjectMapper objectMapper = new com.fasterxml.jackson.databind.ObjectMapper();

        @Override
        public Transaction deserialize(byte[] message) throws IOException {
            return objectMapper.readValue(message, Transaction.class);
        }

        @Override
        public byte[] serialize(Transaction element) {
            try {
                return objectMapper.writeValueAsBytes(element);
            } catch (Exception e) {
                throw new RuntimeException("Failed to serialize transaction", e);
            }
        }

        @Override
        public TypeInformation<Transaction> getProducedType() {
            return TypeInformation.of(Transaction.class);
        }
    }
}

User Profile Model

package com.frauddetection.models;

import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;
import java.io.Serializable;

@Data
@NoArgsConstructor
@AllArgsConstructor
public class UserProfile implements Serializable {
    private String userId;
    private String cardId;
    private double avgTransactionAmount;
    private double stdDevAmount;
    private int totalTransactions;
    private String homeCountry;
    private String homeCity;
    private long lastTransactionTime;
    private boolean isHighRiskUser;

    public void updateWithTransaction(Transaction txn) {
        // Update running statistics
        double oldAvg = this.avgTransactionAmount;
        this.totalTransactions++;

        // Incremental mean
        this.avgTransactionAmount = oldAvg + (txn.getAmount() - oldAvg) / this.totalTransactions;

        // Incremental standard deviation (Welford's method)
        double delta = txn.getAmount() - oldAvg;
        double delta2 = txn.getAmount() - this.avgTransactionAmount;
        this.stdDevAmount = Math.sqrt(
            (Math.pow(this.stdDevAmount, 2) * (this.totalTransactions - 1) + delta * delta2) / this.totalTransactions
        );

        this.lastTransactionTime = txn.getTimestamp();
    }
}

Fraud Alert Model

package com.frauddetection.models;

import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;
import java.io.Serializable;

@Data
@NoArgsConstructor
@AllArgsConstructor
public class FraudAlert implements Serializable {
    private String alertId;
    private String cardId;
    private String userId;
    private String transactionId;
    private FraudType fraudType;
    private double riskScore;
    private String description;
    private long detectedAt;
    private String metadata; // JSON with additional context

    public enum FraudType {
        VELOCITY_FRAUD,          // Too many transactions in short time
        AMOUNT_SPIKE,            // Unusually high amount
        LOCATION_ANOMALY,        // Impossible travel distance
        DECLINING_PATTERN,       // Multiple declines followed by approval
        BEHAVIORAL_ANOMALY,      // ML-detected anomaly
        MERCHANT_RISK,           // High-risk merchant category
        DEVICE_MISMATCH          // Different device for same user
    }
}

Part 3: Core Pipeline Implementation

Main Job (FraudDetectionJob.java)

package com.frauddetection;

import com.frauddetection.models.*;
import com.frauddetection.cep.*;
import com.frauddetection.features.*;
import com.frauddetection.enrichment.*;
import com.frauddetection.ml.*;
import com.frauddetection.sinks.*;

import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.restartstrategy.RestartStrategies;
import org.apache.flink.contrib.streaming.state.RocksDBStateBackend;
import org.apache.flink.runtime.state.StateBackend;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.CheckpointConfig;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer;

import java.time.Duration;
import java.util.Properties;

public class FraudDetectionJob {

    public static void main(String[] args) throws Exception {
        // 1. Setup execution environment
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        // 2. Configure checkpointing
        env.enableCheckpointing(60000); // 1 minute
        CheckpointConfig checkpointConfig = env.getCheckpointConfig();
        checkpointConfig.setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);
        checkpointConfig.setMinPauseBetweenCheckpoints(30000);
        checkpointConfig.setCheckpointTimeout(600000); // 10 minutes
        checkpointConfig.enableExternalizedCheckpoints(
            CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION
        );
        checkpointConfig.enableUnalignedCheckpoints();

        // 3. Configure state backend (RocksDB for large state)
        StateBackend backend = new RocksDBStateBackend("s3://my-bucket/flink/checkpoints", true);
        ((RocksDBStateBackend) backend).enableIncrementalCheckpointing(true);
        env.setStateBackend(backend);

        // 4. Configure restart strategy
        env.setRestartStrategy(RestartStrategies.exponentialDelayRestart(
            Duration.ofSeconds(10),
            Duration.ofMinutes(5),
            2.0,
            Duration.ofHours(1),
            10
        ));

        // 5. Kafka source configuration
        Properties kafkaProps = new Properties();
        kafkaProps.setProperty("bootstrap.servers", "kafka:9092");
        kafkaProps.setProperty("group.id", "fraud-detection");

        FlinkKafkaConsumer<Transaction> kafkaSource = new FlinkKafkaConsumer<>(
            "transactions",
            new Transaction.Schema(),
            kafkaProps
        );

        // 6. Create source stream with watermarks
        DataStream<Transaction> transactions = env
            .addSource(kafkaSource)
            .assignTimestampsAndWatermarks(
                WatermarkStrategy
                    .<Transaction>forBoundedOutOfOrderness(Duration.ofSeconds(10))
                    .withTimestampAssigner((txn, ts) -> txn.getTimestamp())
            )
            .name("Kafka Transaction Source");

        // 7. Enrich transactions with user profiles
        DataStream<EnrichedTransaction> enrichedTransactions = transactions
            .keyBy(Transaction::getCardId)
            .flatMap(new UserProfileEnrichment())
            .name("User Profile Enrichment");

        // 8. Extract ML features
        DataStream<FeatureVector> features = enrichedTransactions
            .keyBy(EnrichedTransaction::getCardId)
            .flatMap(new FeatureExtractor())
            .name("Feature Extraction");

        // 9. CEP-based fraud detection
        DataStream<FraudAlert> velocityAlerts = new VelocityFraudDetector()
            .detect(enrichedTransactions)
            .name("Velocity Fraud Detection");

        DataStream<FraudAlert> locationAlerts = new LocationAnomalyDetector()
            .detect(enrichedTransactions)
            .name("Location Anomaly Detection");

        DataStream<FraudAlert> decliningAlerts = new DecliningPatternDetector()
            .detect(enrichedTransactions)
            .name("Declining Pattern Detection");

        // 10. ML-based fraud scoring
        DataStream<FraudAlert> mlAlerts = features
            .keyBy(FeatureVector::getCardId)
            .flatMap(new MLScoringFunction())
            .name("ML Fraud Scoring");

        // 11. Combine all alerts
        DataStream<FraudAlert> allAlerts = velocityAlerts
            .union(locationAlerts)
            .union(decliningAlerts)
            .union(mlAlerts);

        // 12. Deduplicate and prioritize alerts
        SingleOutputStreamOperator<FraudAlert> deduplicatedAlerts = allAlerts
            .keyBy(FraudAlert::getCardId)
            .flatMap(new AlertDeduplicator())
            .name("Alert Deduplication");

        // 13. Sink to Kafka (for downstream consumers)
        deduplicatedAlerts
            .addSink(new FlinkKafkaProducer<>(
                "fraud-alerts",
                new FraudAlert.Schema(),
                kafkaProps
            ))
            .name("Kafka Alert Sink");

        // 14. Sink to MySQL (for dashboard/analysis)
        deduplicatedAlerts
            .addSink(new AlertSink())
            .name("MySQL Alert Sink");

        // 15. Side output high-risk alerts to PagerDuty
        deduplicatedAlerts
            .filter(alert -> alert.getRiskScore() > 0.9)
            .addSink(new PagerDutySink())
            .name("PagerDuty High-Risk Alerts");

        // 16. Execute job
        env.execute("Real-Time Fraud Detection");
    }
}

Part 4: CEP Fraud Detectors

Velocity Fraud Detector

package com.frauddetection.cep;

import com.frauddetection.models.*;
import org.apache.flink.cep.CEP;
import org.apache.flink.cep.PatternStream;
import org.apache.flink.cep.pattern.Pattern;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.windowing.time.Time;

public class VelocityFraudDetector {

    public DataStream<FraudAlert> detect(DataStream<EnrichedTransaction> stream) {
        // Pattern: >5 transactions in 10 minutes totaling >$10K
        Pattern<EnrichedTransaction, ?> pattern = Pattern
            .<EnrichedTransaction>begin("rapid_txns")
            .where(txn -> txn.getTransaction().getAmount() > 100)
            .timesOrMore(5)
            .within(Time.minutes(10));

        PatternStream<EnrichedTransaction> patternStream = CEP.pattern(
            stream.keyBy(et -> et.getTransaction().getCardId()),
            pattern
        );

        return patternStream.select((map) -> {
            List<EnrichedTransaction> txns = map.get("rapid_txns");
            double totalAmount = txns.stream()
                .mapToDouble(et -> et.getTransaction().getAmount())
                .sum();

            EnrichedTransaction first = txns.get(0);
            EnrichedTransaction last = txns.get(txns.size() - 1);

            if (txns.size() >= 5 && totalAmount > 10000) {
                double riskScore = calculateRiskScore(txns, totalAmount);

                return new FraudAlert(
                    UUID.randomUUID().toString(),
                    first.getTransaction().getCardId(),
                    first.getTransaction().getUserId(),
                    last.getTransaction().getTransactionId(),
                    FraudAlert.FraudType.VELOCITY_FRAUD,
                    riskScore,
                    String.format("Velocity fraud: %d transactions ($%.2f) in 10 min",
                        txns.size(), totalAmount),
                    System.currentTimeMillis(),
                    buildMetadata(txns)
                );
            }
            return null;
        }).filter(Objects::nonNull);
    }

    private double calculateRiskScore(List<EnrichedTransaction> txns, double totalAmount) {
        // Risk factors:
        // 1. Number of transactions (weight: 0.3)
        // 2. Total amount (weight: 0.4)
        // 3. Deviation from user average (weight: 0.3)

        double txnScore = Math.min(txns.size() / 10.0, 1.0); // Max at 10 txns
        double amountScore = Math.min(totalAmount / 20000.0, 1.0); // Max at $20K

        EnrichedTransaction first = txns.get(0);
        double avgAmount = first.getUserProfile().getAvgTransactionAmount();
        double deviationScore = avgAmount > 0 ? Math.min(totalAmount / (avgAmount * 5), 1.0) : 0.5;

        return 0.3 * txnScore + 0.4 * amountScore + 0.3 * deviationScore;
    }

    private String buildMetadata(List<EnrichedTransaction> txns) {
        // Build JSON metadata
        Map<String, Object> metadata = new HashMap<>();
        metadata.put("transaction_count", txns.size());
        metadata.put("total_amount", txns.stream().mapToDouble(et -> et.getTransaction().getAmount()).sum());
        metadata.put("transaction_ids", txns.stream().map(et -> et.getTransaction().getTransactionId()).collect(Collectors.toList()));

        try {
            return new ObjectMapper().writeValueAsString(metadata);
        } catch (Exception e) {
            return "{}";
        }
    }
}

Location Anomaly Detector

package com.frauddetection.cep;

import com.frauddetection.models.*;
import org.apache.flink.cep.CEP;
import org.apache.flink.cep.pattern.Pattern;
import org.apache.flink.cep.pattern.conditions.IterativeCondition;

public class LocationAnomalyDetector {

    public DataStream<FraudAlert> detect(DataStream<EnrichedTransaction> stream) {
        // Pattern: Two transactions from different locations with impossible travel speed
        Pattern<EnrichedTransaction, ?> pattern = Pattern
            .<EnrichedTransaction>begin("first")
            .where(et -> true)
            .next("second")
            .where(new IterativeCondition<EnrichedTransaction>() {
                @Override
                public boolean filter(EnrichedTransaction current, Context<EnrichedTransaction> ctx) throws Exception {
                    EnrichedTransaction first = ctx.getEventsForPattern("first").iterator().next();

                    // Calculate distance and time
                    double distance = calculateDistance(
                        first.getTransaction().getLatitude(),
                        first.getTransaction().getLongitude(),
                        current.getTransaction().getLatitude(),
                        current.getTransaction().getLongitude()
                    );

                    long timeDiffMs = current.getTransaction().getTimestamp() - first.getTransaction().getTimestamp();
                    double timeDiffHours = timeDiffMs / 3600000.0;

                    // Calculate speed (km/h)
                    double speedKmh = distance / timeDiffHours;

                    // Flag if speed > 800 km/h (requires flight, suspicious for short time)
                    return speedKmh > 800 && timeDiffHours < 6;
                }
            })
            .within(Time.hours(6));

        PatternStream<EnrichedTransaction> patternStream = CEP.pattern(
            stream.keyBy(et -> et.getTransaction().getCardId()),
            pattern
        );

        return patternStream.select((map) -> {
            EnrichedTransaction first = map.get("first").get(0);
            EnrichedTransaction second = map.get("second").get(0);

            double distance = calculateDistance(
                first.getTransaction().getLatitude(),
                first.getTransaction().getLongitude(),
                second.getTransaction().getLatitude(),
                second.getTransaction().getLongitude()
            );

            long timeDiffMs = second.getTransaction().getTimestamp() - first.getTransaction().getTimestamp();
            double speedKmh = (distance / (timeDiffMs / 3600000.0));

            return new FraudAlert(
                UUID.randomUUID().toString(),
                second.getTransaction().getCardId(),
                second.getTransaction().getUserId(),
                second.getTransaction().getTransactionId(),
                FraudAlert.FraudType.LOCATION_ANOMALY,
                0.85, // High risk
                String.format("Impossible travel: %.0f km in %.1f hours (%.0f km/h)",
                    distance, timeDiffMs / 3600000.0, speedKmh),
                System.currentTimeMillis(),
                String.format("{\"distance_km\": %.2f, \"speed_kmh\": %.2f}", distance, speedKmh)
            );
        });
    }

    private double calculateDistance(double lat1, double lon1, double lat2, double lon2) {
        // Haversine formula
        final int R = 6371; // Radius of Earth in km

        double latDistance = Math.toRadians(lat2 - lat1);
        double lonDistance = Math.toRadians(lon2 - lon1);

        double a = Math.sin(latDistance / 2) * Math.sin(latDistance / 2)
                + Math.cos(Math.toRadians(lat1)) * Math.cos(Math.toRadians(lat2))
                * Math.sin(lonDistance / 2) * Math.sin(lonDistance / 2);

        double c = 2 * Math.atan2(Math.sqrt(a), Math.sqrt(1 - a));

        return R * c;
    }
}

Declining Pattern Detector

package com.frauddetection.cep;

public class DecliningPatternDetector {

    public DataStream<FraudAlert> detect(DataStream<EnrichedTransaction> stream) {
        // Pattern: 3 declined transactions followed by 1 approved high-value transaction
        Pattern<EnrichedTransaction, ?> pattern = Pattern
            .<EnrichedTransaction>begin("declines")
            .where(et -> et.getTransaction().getStatus().equals("DECLINED"))
            .times(3)
            .followedBy("approve")
            .where(et -> et.getTransaction().getStatus().equals("APPROVED")
                && et.getTransaction().getAmount() > 500)
            .within(Time.minutes(15));

        PatternStream<EnrichedTransaction> patternStream = CEP.pattern(
            stream.keyBy(et -> et.getTransaction().getCardId()),
            pattern
        );

        return patternStream.select((map) -> {
            List<EnrichedTransaction> declines = map.get("declines");
            EnrichedTransaction approve = map.get("approve").get(0);

            return new FraudAlert(
                UUID.randomUUID().toString(),
                approve.getTransaction().getCardId(),
                approve.getTransaction().getUserId(),
                approve.getTransaction().getTransactionId(),
                FraudAlert.FraudType.DECLINING_PATTERN,
                0.92, // Very high risk
                String.format("Credential testing: %d declines followed by $%.2f approval",
                    declines.size(), approve.getTransaction().getAmount()),
                System.currentTimeMillis(),
                buildDeclineMetadata(declines, approve)
            );
        });
    }
}

Part 5: Feature Engineering & ML Scoring

Feature Extractor

package com.frauddetection.features;

import com.frauddetection.models.*;
import org.apache.flink.api.common.functions.RichFlatMapFunction;
import org.apache.flink.api.common.state.*;
import org.apache.flink.configuration.Configuration;

public class FeatureExtractor extends RichFlatMapFunction<EnrichedTransaction, FeatureVector> {

    private transient MapState<String, Double> features;
    private transient ListState<Transaction> recentTransactions;

    @Override
    public void open(Configuration config) {
        // Feature state
        MapStateDescriptor<String, Double> featureDesc =
            new MapStateDescriptor<>("features", String.class, Double.class);
        features = getRuntimeContext().getMapState(featureDesc);

        // Recent transactions (for velocity features)
        ListStateDescriptor<Transaction> txnDesc =
            new ListStateDescriptor<>("recent_txns", Transaction.class);
        txnDesc.enableTimeToLive(
            StateTtlConfig.newBuilder(Time.hours(24)).build()
        );
        recentTransactions = getRuntimeContext().getListState(txnDesc);
    }

    @Override
    public void flatMap(EnrichedTransaction enriched, Collector<FeatureVector> out) throws Exception {
        Transaction txn = enriched.getTransaction();
        UserProfile profile = enriched.getUserProfile();

        // Feature 1: Velocity features (last 1 hour)
        List<Transaction> recent = new ArrayList<>();
        long oneHourAgo = txn.getTimestamp() - 3600000;
        for (Transaction t : recentTransactions.get()) {
            if (t.getTimestamp() > oneHourAgo) {
                recent.add(t);
            }
        }
        recentTransactions.add(txn);

        features.put("txn_count_1h", (double) recent.size());
        features.put("total_amount_1h", recent.stream().mapToDouble(Transaction::getAmount).sum());

        // Feature 2: Behavioral features
        features.put("amount", txn.getAmount());
        features.put("hour_of_day", (double) (txn.getTimestamp() % 86400000) / 3600000);
        features.put("day_of_week", (double) ((txn.getTimestamp() / 86400000) % 7));

        // Feature 3: Historical features from profile
        features.put("avg_amount", profile.getAvgTransactionAmount());
        features.put("stddev_amount", profile.getStdDevAmount());
        features.put("total_txns", (double) profile.getTotalTransactions());

        // Feature 4: Amount deviation
        double zScore = profile.getStdDevAmount() > 0
            ? (txn.getAmount() - profile.getAvgTransactionAmount()) / profile.getStdDevAmount()
            : 0.0;
        features.put("amount_zscore", zScore);

        // Feature 5: Merchant category risk (simplified)
        features.put("merchant_risk", calculateMerchantRisk(txn.getMerchantCategory()));

        // Feature 6: Time since last transaction
        long timeSinceLastMs = txn.getTimestamp() - profile.getLastTransactionTime();
        features.put("time_since_last_txn_hours", timeSinceLastMs / 3600000.0);

        // Feature 7: International transaction flag
        features.put("is_international", txn.getCountry().equals(profile.getHomeCountry()) ? 0.0 : 1.0);

        // Feature 8: Unique merchants in last hour
        long uniqueMerchants = recent.stream()
            .map(Transaction::getMerchantId)
            .distinct()
            .count();
        features.put("unique_merchants_1h", (double) uniqueMerchants);

        // Create feature vector
        Map<String, Double> featureMap = new HashMap<>();
        for (Map.Entry<String, Double> entry : features.entries()) {
            featureMap.put(entry.getKey(), entry.getValue());
        }

        out.collect(new FeatureVector(
            txn.getCardId(),
            txn.getUserId(),
            txn.getTransactionId(),
            txn.getTimestamp(),
            featureMap
        ));
    }

    private double calculateMerchantRisk(String category) {
        // Simplified risk mapping
        switch (category.toLowerCase()) {
            case "gambling":
            case "cryptocurrency":
                return 0.9;
            case "travel":
            case "electronics":
                return 0.6;
            case "groceries":
            case "utilities":
                return 0.1;
            default:
                return 0.5;
        }
    }
}

ML Scoring Function

package com.frauddetection.ml;

import com.frauddetection.models.*;

public class MLScoringFunction extends RichFlatMapFunction<FeatureVector, FraudAlert> {

    private transient MLModel model;

    @Override
    public void open(Configuration config) throws Exception {
        // Load pre-trained model (e.g., from S3 or model registry)
        model = MLModel.load("s3://models/fraud-detection-v1.model");
    }

    @Override
    public void flatMap(FeatureVector features, Collector<FraudAlert> out) throws Exception {
        // Score using ML model
        double fraudProbability = model.predict(features.getFeatures());

        // Threshold: Alert if probability > 0.7
        if (fraudProbability > 0.7) {
            out.collect(new FraudAlert(
                UUID.randomUUID().toString(),
                features.getCardId(),
                features.getUserId(),
                features.getTransactionId(),
                FraudAlert.FraudType.BEHAVIORAL_ANOMALY,
                fraudProbability,
                String.format("ML-detected anomaly (score: %.2f)", fraudProbability),
                System.currentTimeMillis(),
                String.format("{\"model_version\": \"v1\", \"features\": %s}",
                    new ObjectMapper().writeValueAsString(features.getFeatures()))
            ));
        }
    }
}

// Simplified ML Model wrapper
class MLModel {
    private Map<String, Double> weights;

    public static MLModel load(String path) {
        // In production: Load from S3, DynamoDB, or model registry
        MLModel model = new MLModel();
        model.weights = new HashMap<>();
        // Feature weights (trained offline)
        model.weights.put("amount_zscore", 0.25);
        model.weights.put("txn_count_1h", 0.20);
        model.weights.put("merchant_risk", 0.15);
        model.weights.put("is_international", 0.15);
        model.weights.put("time_since_last_txn_hours", -0.10);
        model.weights.put("unique_merchants_1h", 0.10);
        return model;
    }

    public double predict(Map<String, Double> features) {
        // Simple linear model: score = sigmoid(sum(weight_i * feature_i))
        double rawScore = 0.0;
        for (Map.Entry<String, Double> entry : weights.entrySet()) {
            rawScore += weights.getOrDefault(entry.getKey(), 0.0) * features.getOrDefault(entry.getKey(), 0.0);
        }
        // Sigmoid
        return 1.0 / (1.0 + Math.exp(-rawScore));
    }
}

Part 6: Deployment & Monitoring

Kubernetes Deployment

# k8s/flink-fraud-detection.yaml
apiVersion: v1
kind: ConfigMap
metadata:
  name: flink-fraud-config
  namespace: fraud-detection
data:
  flink-conf.yaml: |
    jobmanager.rpc.address: flink-jobmanager
    taskmanager.numberOfTaskSlots: 8
    parallelism.default: 32
    state.backend: rocksdb
    state.checkpoints.dir: s3://fraud-detection/checkpoints
    state.savepoints.dir: s3://fraud-detection/savepoints
    high-availability: kubernetes
    high-availability.storageDir: s3://fraud-detection/ha
    metrics.reporter.prom.class: org.apache.flink.metrics.prometheus.PrometheusReporter
    metrics.reporter.prom.port: 9249
    taskmanager.memory.process.size: 16384m
    taskmanager.memory.managed.fraction: 0.4
---
apiVersion: apps/v1
kind: StatefulSet
metadata:
  name: flink-jobmanager
  namespace: fraud-detection
spec:
  serviceName: flink-jobmanager
  replicas: 2
  selector:
    matchLabels:
      app: flink
      component: jobmanager
  template:
    metadata:
      labels:
        app: flink
        component: jobmanager
    spec:
      containers:
      - name: jobmanager
        image: fraud-detection-flink:1.0
        args: ["jobmanager"]
        resources:
          requests:
            memory: "4Gi"
            cpu: "2"
          limits:
            memory: "4Gi"
            cpu: "4"
---
apiVersion: apps/v1
kind: Deployment
metadata:
  name: flink-taskmanager
  namespace: fraud-detection
spec:
  replicas: 10
  selector:
    matchLabels:
      app: flink
      component: taskmanager
  template:
    metadata:
      labels:
        app: flink
        component: taskmanager
    spec:
      containers:
      - name: taskmanager
        image: fraud-detection-flink:1.0
        args: ["taskmanager"]
        resources:
          requests:
            memory: "16Gi"
            cpu: "4"
          limits:
            memory: "16Gi"
            cpu: "8"

Prometheus Alerts

# monitoring/alerts.yaml
groups:
- name: fraud_detection
  rules:
  - alert: HighFraudDetectionLatency
    expr: flink_taskmanager_job_task_operator_latency_p99 > 5000
    for: 10m
    annotations:
      summary: "Fraud detection latency > 5s"

  - alert: CheckpointFailure
    expr: flink_jobmanager_job_numberOfFailedCheckpoints > 3
    for: 15m
    annotations:
      summary: "Multiple checkpoint failures"

  - alert: HighBackpressure
    expr: flink_taskmanager_job_task_backPressureTimeMsPerSecond > 800
    for: 20m
    annotations:
      summary: "Sustained backpressure detected"

  - alert: AlertSinkLag
    expr: flink_taskmanager_job_task_operator_numRecordsOutPerSecond{operator_name="Alert Sink"} < 100
    for: 10m
    annotations:
      summary: "Alert sink throughput dropped"

Part 7: Testing & Validation

Unit Tests

package com.frauddetection;

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.test.util.MiniClusterWithClientResource;
import org.junit.ClassRule;
import org.junit.Test;

public class FraudDetectionTest {

    @ClassRule
    public static MiniClusterWithClientResource flinkCluster =
        new MiniClusterWithClientResource(
            new MiniClusterResourceConfiguration.Builder()
                .setNumberSlotsPerTaskManager(4)
                .setNumberTaskManagers(1)
                .build());

    @Test
    public void testVelocityFraudDetection() throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        // Create test transactions
        DataStream<Transaction> testStream = env.fromElements(
            new Transaction("txn1", "card1", "user1", 100, "m1", "retail", "US", "NYC", 40.7, -74.0, "APPROVED", 1000L, "dev1", "1.1.1.1"),
            new Transaction("txn2", "card1", "user1", 200, "m2", "retail", "US", "NYC", 40.7, -74.0, "APPROVED", 2000L, "dev1", "1.1.1.1"),
            new Transaction("txn3", "card1", "user1", 300, "m3", "retail", "US", "NYC", 40.7, -74.0, "APPROVED", 3000L, "dev1", "1.1.1.1"),
            new Transaction("txn4", "card1", "user1", 400, "m4", "retail", "US", "NYC", 40.7, -74.0, "APPROVED", 4000L, "dev1", "1.1.1.1"),
            new Transaction("txn5", "card1", "user1", 500, "m5", "retail", "US", "NYC", 40.7, -74.0, "APPROVED", 5000L, "dev1", "1.1.1.1"),
            new Transaction("txn6", "card1", "user1", 600, "m6", "retail", "US", "NYC", 40.7, -74.0, "APPROVED", 6000L, "dev1", "1.1.1.1") // Should trigger
        );

        // Apply velocity detector
        DataStream<FraudAlert> alerts = new VelocityFraudDetector()
            .detect(testStream.map(txn -> new EnrichedTransaction(txn, new UserProfile())));

        // Collect results
        List<FraudAlert> results = alerts.executeAndCollect(10);

        // Verify
        assertEquals(1, results.size());
        assertEquals(FraudAlert.FraudType.VELOCITY_FRAUD, results.get(0).getFraudType());
    }
}

Summary & Next Steps

What You’ve Built

A production-grade fraud detection system with: ✅ Real-time processing: Sub-second latency for millions of transactions ✅ Multi-strategy detection: CEP patterns + ML scoring ✅ Stateful processing: User profiles, transaction history ✅ Fault tolerance: Checkpointing, HA, exactly-once semantics ✅ Scalability: Horizontal scaling with Kubernetes ✅ Observability: Prometheus metrics, Grafana dashboards, PagerDuty alerts

Production Checklist

Enhancements

Advanced ML: Train XGBoost/LightGBM models, A/B testing
Graph Analysis: Neo4j for transaction networks
Real-time Model Updates: Online learning, model versioning
Multi-Region Deployment: Active-active for global coverage
Cost Optimization: Spot instances, auto-scaling

Resources

Documentation

Papers

“State Management in Apache Flink”

Congratulations! You’ve mastered Apache Flink by building a production-grade system. This project demonstrates all core Flink concepts: DataStream API, CEP, state management, checkpointing, and production deployment. You’re now ready to architect and deploy real-time streaming applications at scale!

Overview

Testing & Code Quality

Crash Courses

AI Engineering

Math for ML - Understanding Linear Algebra

Probability & Statistics for ML

Math for ML - Understanding Calculus

ML Mastery

Deep Learning Mastery

NestJS Mastery

Microservices Mastery

Low Level Design

OOP Concepts

SOLID Principles

Design Patterns

LLD Case Studies

System Design (HLD)

Senior Level (L5+/Staff)

HLD Case Studies

Engineering Fundamentals

DevOps & Operations

Azure Cloud Engineering

AWS Cloud

AWS Monitoring & Observability

AWS Security Services

AWS Serverless

AWS Operations

AWS Advanced

AWS Case Studies

GCP Cloud Engineering

DevOps Tools

Database Engineering

HIPAA Compliance Mastery

Operating Systems

Linux Internals

Distributed Systems

Networking Mastery

Build Your Own X

Go Lang Mastery

C Programming

Classic Research Papers

Distributed System Tools

​Capstone: Production-Grade Real-Time Fraud Detection System

​Project Overview

​What You’ll Build

​System Architecture

​Part 1: Project Setup

​Prerequisites

​Project Structure

​Maven Dependencies (pom.xml)

​Part 2: Data Models

​Transaction Model

​User Profile Model

​Fraud Alert Model

​Part 3: Core Pipeline Implementation

​Main Job (FraudDetectionJob.java)

​Part 4: CEP Fraud Detectors

​Velocity Fraud Detector

​Location Anomaly Detector

​Declining Pattern Detector

​Part 5: Feature Engineering & ML Scoring

​Feature Extractor

​ML Scoring Function

​Part 6: Deployment & Monitoring

​Kubernetes Deployment

​Prometheus Alerts

​Part 7: Testing & Validation

​Unit Tests

​Summary & Next Steps

​What You’ve Built

​Production Checklist

​Enhancements

​Resources

​Documentation

​Papers

Capstone: Production-Grade Real-Time Fraud Detection System

Project Overview

What You’ll Build

System Architecture

Part 1: Project Setup

Prerequisites

Project Structure

Maven Dependencies (pom.xml)

Part 2: Data Models

Transaction Model

User Profile Model

Fraud Alert Model

Part 3: Core Pipeline Implementation

Main Job (FraudDetectionJob.java)

Part 4: CEP Fraud Detectors

Velocity Fraud Detector

Location Anomaly Detector

Declining Pattern Detector

Part 5: Feature Engineering & ML Scoring

Feature Extractor

ML Scoring Function

Part 6: Deployment & Monitoring

Kubernetes Deployment

Prometheus Alerts

Part 7: Testing & Validation

Unit Tests

Summary & Next Steps

What You’ve Built

Production Checklist

Enhancements

Resources

Documentation

Papers