Real-Time Fraud Detection with a Streaming Machine Learning Pipeline

This project implements a complete, end-to-end streaming machine learning pipeline for real-time fraud detection. It ingests a stream of mock financial transactions using Kafka, which are then processed in near real-time by an Apache Spark job and stored in an S3 data lake using the Apache Hudi format. Feature engineering and model training are orchestrated by Apache Airflow, which uses Feast to serve the latest features to an online Redis store and MLflow to track experiments and register the trained model. Finally, predictions are served in real-time through a FastAPI backend and visualized in a Streamlit web interface.

The entire infrastructure is defined as code (IaC) using Terraform and deployed on kubernetes Clusters using Amazon EKS via a GitOps workflow with ArgoCD.

Architecture Diagram

The diagram below illustrates the flow of data and interactions between the various components of the system.

graph TD
    %% Style Definitions for Colors
    classDef dataStyle fill:#e6f3ff,stroke:#333,stroke-width:2px,color:#333
    classDef computeStyle fill:#fff4e6,stroke:#333,stroke-width:2px,color:#333
    classDef mlStyle fill:#e6ffed,stroke:#333,stroke-width:2px,color:#333
    classDef serveStyle fill:#e6f9ff,stroke:#333,stroke-width:2px,color:#333
    classDef gitopsStyle fill:#f2e6ff,stroke:#333,stroke-width:2px,color:#333

    subgraph "Data Ingestion & Processing"
        A["fa:fa-paper-plane Event Producer"] -- Transactions (JSON) --> B["fa:fa-stream Kafka Topic"];
        B -- Stream --> C["fa:fa-bolt Spark Streaming Job"];
        C -- Processes & Aggregates --> D["fa:fa-database S3 Offline Store (Hudi)"];
    end

    subgraph "Feature & ML Platform"
        E["fa:fa-cogs Airflow"] -- Triggers Daily --> F["fa:fa-sync Feature Materialization"];
        F -- Reads from Hudi --> D;
        F -- Writes Online Features --> G["fa:fa-database-circle-plus Redis Online Store"];
        E -- Triggers --> H["fa:fa-chart-line Model Training"];
        H -- Reads Historical Features --> D;
        H -- Logs Experiments & Models --> I["fa:fa-flask MLflow Tracking Server"];
        I -- Stores Artifacts --> J["fa:fa-archive S3 MLflow Artifacts"];
    end

    subgraph "Real-Time Serving"
        K["fa:fa-desktop Streamlit Frontend"] -- HTTP Request --> L["fa:fa-server Inference API (FastAPI)"];
        L -- Gets Latest Features --> G;
        L -- Loads Model --> I;
        L -- Returns Prediction --> K;
    end

    subgraph "Kubernetes & GitOps"
        M["fa:fa-github GitHub Repo"] -- Source of Truth --> N["fa:fa-sync-alt ArgoCD"];
        N -- Deploys & Manages --> O["fa:fa-cubes Amazon EKS Cluster"];
        O --> P[Deployed Services];
    end

    %% Apply Styles to Nodes
    class A,C,F,H computeStyle
    class B,D,G,J dataStyle
    class E,I mlStyle
    class K,L,P serveStyle
    class M,N,O gitopsStyle

Loading

Technology Stack

• Infrastructure: AWS (EKS, S3, IAM), Terraform
• Streaming: Apache Kafka (Strimzi Operator), Apache Spark (Spark on K8s Operator)
• Data Lake Format: Apache Hudi
• Feature Store: Feast
• CI/CD: Github-Actions, ArgoCD
• ML Orchestration: MLflow, Apache Airflow
• Online Datastore: Redis
• ML Algorithm: XGBoost
• Serving: FastAPI, Streamlit
• Deployment: Kubernetes, Docker, ArgoCD, Kustomize

Prerequisites

• AWS CLI
• Terraform
• kubectl
• argocd-cli
• Docker
• A GitHub account and a Personal Access Token (PAT) with repo scope.

1. Infrastructure Setup (Terraform)

The core cloud infrastructure (EKS Cluster, S3 Buckets, IAM Roles) is managed by Terraform.

1. Navigate to the Terraform directory:

   cd infra/terraform/environments/dev

2. Initialize Terraform:

   terraform init

3. Deploy the infrastructure. This will take several minutes.

   terraform apply

4. After the apply is complete, update your local kubeconfig to connect to the new EKS cluster:

   aws eks update-kubeconfig --name <YOUR_EKS_CLUSTER_NAME> --region <YOUR_AWS_REGION>

2. Kubernetes Setup (ArgoCD)

We will use ArgoCD to manage all applications running on the cluster.

1. Install ArgoCD into the cluster:

   kubectl create namespace argocd
   kubectl apply -n argocd -f https://raw.githubusercontent.com/argoproj/argo-cd/stable/manifests/install.yaml

2. Expose the ArgoCD server with a LoadBalancer to access its UI:

   kubectl patch svc argocd-server -n argocd -p '{"spec":{"type":"LoadBalancer"}}'

3. Get the ArgoCD server URL and initial password:

   # Get URL
   kubectl get svc argocd-server -n argocd -o jsonpath='{.status.loadBalancer.ingress[0].hostname}'
   
   # Get initial password
   kubectl -n argocd get secret argocd-initial-admin-secret -o jsonpath="{.data.password}" | base64 -d

4. Log in to ArgoCD via the CLI and add your Git repository:

   argocd login <ARGOCD_SERVER_URL>
   
   argocd repo add https://github.com/Nonzzo/streaming-ml-pipeline.git \
     --username <YOUR_GITHUB_USERNAME> \
     --password <YOUR_GITHUB_PAT>
   

3. Application Deployment (GitOps)

Deploy the root ArgoCD application, which will in turn deploy all other services defined in the infra/k8s/apps/dev directory.

kubectl apply -f infra/k8s/root-app/dev/root-app.yaml

ArgoCD will now automatically sync the state defined in your Git repository with the EKS cluster. You can monitor the progress in the ArgoCD UI.

3. Application Deployment (GitOps)

Deploy the root ArgoCD application, which will in turn deploy all other services defined in the infra/k8s/apps/dev directory.

kubectl apply -f infra/k8s/root-app/dev/root-app.yaml

ArgoCD will now automatically sync the state defined in your Git repository with the EKS cluster. You can monitor the progress in the ArgoCD UI.

4. Install Strimzi operator in the data namespace

kubectl create namespace data kubectl create -f 'https://strimzi.io/install/latest?namespace=data' -n data