Real-Time Fraud Detection on Azure

Introduction

30 Second Summary

Every time you tap your card at a coffee shop, your bank has milliseconds to decide if that transaction is legitimate or fraudulent. Behind the scenes, machine learning models score each transaction in real time to protect your money.

In this project, you will build a complete real-time fraud detection pipeline on Azure. You'll stream simulated transactions through a medallion data architecture, train a machine learning model, and deploy an inference API that scores transactions as they arrive.

What You'll Build

A production-grade fraud detection system that streams transactions through Azure Event Hubs, processes them with Spark Structured Streaming on Azure Databricks, trains an XGBoost model tracked by MLflow, and serves predictions via a FastAPI endpoint on Azure Container Apps.

By the end of this project, you'll have:

• A Kafka-compatible transaction simulator streaming data to Azure Event Hubs, with a Spark pipeline processing transactions through Bronze, Silver, and Gold layers on Azure Databricks.
• A trained XGBoost fraud detection model registered in Azure ML with full MLflow experiment tracking.
• A live FastAPI inference endpoint on Azure Container Apps that scores transactions using features served from Redis.
• Secret Mission: A real-time alerting system that flags high-confidence fraud predictions and sends webhook notifications.

Are there any prerequisites?

No prior experience with Azure or machine learning is required. This guide walks you through setting up every tool from scratch, including your Azure account, Python environment, Docker, and all cloud resources.

Set Up Your Azure Account and Local Tools

Every component of your fraud detection pipeline relies on Python scripts and Azure cloud services. Before you can stream transactions, train models, or deploy APIs, you need three foundations in place: a Python runtime, the Azure CLI for managing cloud resources, and an active Azure subscription.

In this step, you will install each of these foundations and verify they work. Once complete, you will be ready to start building infrastructure in Azure.

In this step, get ready to:

• Install Python 3.11+ and verify it from the command line.
• Install the Azure CLI using WinGet and verify with az version.
• Create a free Azure account and log in via az login.

Install Python 3.11+

This project requires Python version 3.11 or higher. Several dependencies (including the ML libraries you will use later) need features introduced in 3.11.

• Press the Windows key to open your search bar.
• Type Terminal and press Enter to open Windows Terminal.
• Check if Python is already installed by running this command:

python --version

Look at the output and pick the tab below that matches what you see.

✔️ I see version 3.11 or higher

Your output shows something like Python 3.11.x or Python 3.12.x. Python is ready. Skip ahead to the next substep.

ⓧ I see an older version

Your output shows a version below 3.11 (for example Python 3.9.x or Python 3.10.x). You need to install a newer version.

• Open https://www.python.org/downloads/ in your browser.
• Click the large Download Python 3.1x.x button (the latest stable version 3.11 or above).
• Run the installer. On the very first screen, check the box labeled Add python.exe to PATH before clicking Install Now.
• Close and reopen Windows Terminal, then run python --version again. You should now see 3.11 or higher.

Why reopen the terminal?

The PATH environment variable is loaded when a terminal session starts. Closing and reopening the terminal picks up the new path that the Python installer added.

ⓧ 'python' is not recognized

This means Python is not installed on your machine yet.

• Open https://www.python.org/downloads/ in your browser.
• Click the large Download Python 3.1x.x button (the latest stable version 3.11 or above).
• Run the installer. On the very first screen, check the box labeled Add python.exe to PATH before clicking Install Now.
• Close and reopen Windows Terminal, then run python --version again. You should now see 3.11 or higher.

Add python.exe to PATH

This checkbox is critical. Without it, Windows will not find the python command in your terminal. If you missed it, re-run the installer, select Modify, and enable the PATH option.

Still seeing an older version after reinstalling?

If you installed a newer Python but python --version still reports the old version, the old install is earlier in your PATH. Uninstall the old version from Settings > Apps > Installed apps, then reopen your terminal and check again.

Need more help? Help me fix my Python version on Windows

Install the Azure CLI via WinGet

The Azure CLI is a command-line tool that lets you create and manage Azure resources directly from your terminal. You will use it throughout this project to provision cloud services.

• Install the Azure CLI by running this command in Windows Terminal:

winget install --exact --id Microsoft.AzureCLI

What does this command do?

WinGet is the built-in Windows package manager. The --exact --id flags ensure you install the official Microsoft Azure CLI package and not a similarly named one.

• The installer may prompt you to agree to source agreements. Type Y and press Enter to accept.
• Once installation finishes, close and reopen Windows Terminal so the new PATH entry loads.
• Verify the installation by running:

az version

You should see JSON output showing the CLI version (2.87.0 or higher) and extension versions.

✔️ I see a version number

Your output includes a line like "azure-cli": "2.87.0" or newer. The Azure CLI is ready. Proceed to the next substep.

ⓧ 'az' is not recognized

Windows cannot find the az command. This usually means the terminal session loaded before the install finished writing PATH.

• Make sure the WinGet install completed without errors.
• Close all terminal windows and open a fresh Windows Terminal.
• Run az version again.

If the issue persists, winget may not be available on your system. Download the Azure CLI installer directly from https://learn.microsoft.com/en-us/cli/azure/install-azure-cli-windows and run the MSI.

WinGet itself not found?

WinGet comes pre-installed on Windows 10 (version 1809+) and Windows 11. If your system does not have it, open the Microsoft Store, search for App Installer, and install or update it. Then retry the command above.

Still stuck? Help me install the Azure CLI on Windows

Create your Azure free account and log in

Azure's free account gives you $200 in credit valid for 30 days. That is more than enough to complete this project if you tear down resources after each session.

• Open https://azure.microsoft.com/en-us/pricing/purchase-options/azure-account in your browser.
• Click Try Azure for free.
• Sign in with a Microsoft account or create one.
• Complete the verification steps. You will need a phone number and a non-prepaid credit or debit card (you will not be charged).
• Once your account is confirmed, return to Windows Terminal.

Why do I need a credit card?

Microsoft uses the card for identity verification only. You will not be charged unless you manually upgrade to pay-as-you-go after your free credit runs out.

• Log in to Azure from the CLI by running:

az login

What happens when you run az login?

On Windows 10+, the Azure CLI uses Web Account Manager (WAM) to open a sign-in dialog. Sign in with the same Microsoft account you used to create your free Azure account.

If a sign-in dialog does not appear, the CLI will fall back to opening your browser instead. Follow the prompts displayed in your terminal.

• After signing in, the terminal will display JSON output listing your subscriptions. Look for a line showing your free trial subscription with a status of Enabled.

✔️ I see my subscription listed

You are logged in and your Azure subscription is active. You are ready to start creating cloud resources.

ⓧ I see an authentication error

Azure now requires multifactor authentication (MFA) for CLI sign-in. Make sure you complete any MFA prompts (authenticator app, SMS code) that appear.

• If the sign-in window did not appear at all, try running:

az login --use-device-code

This displays a code and a URL. Open the URL in any browser, enter the code, and complete the sign-in there.

Still stuck? Help me troubleshoot az login on Windows

Your local tools are installed and your Azure subscription is live. Next up, you will install Docker Desktop and scaffold your Python project folder with all its dependencies.

Install Docker and Set Up Your Python Project

In the last step, you got Python and the Azure CLI ready. Now it's time to install the remaining tools and create the project structure that will hold all of your pipeline code.

Later in this project, you'll package your fraud detection API into a container for deployment on Azure. Docker makes that possible by letting you build and run containers locally before pushing them to the cloud.

In this step, get ready to:

• Install Docker Desktop and verify it's running.
• Install VS Code for editing your project files.
• Create the project folder structure with a Python virtual environment and all dependencies.

Install Docker Desktop

Docker Desktop is the application that lets you build and run containers on your local machine. You'll need it later to containerize your FastAPI inference API before deploying it to Azure.

• Open your browser and go to the Docker Desktop download page:

https://www.docker.com/products/docker-desktop/

• Download the installer for your operating system (Windows or macOS).
• Run the installer and follow the prompts to complete the installation.
• Once the installation finishes, open Docker Desktop.

What happens during Docker Desktop setup?

On Windows, the installer may ask you to enable WSL 2 (Windows Subsystem for Linux). Accept this. Docker Desktop uses WSL 2 as its default backend for running Linux containers.

On macOS, you may be prompted to grant system permissions. Allow these so Docker can manage containers.

• Wait for Docker Desktop to finish starting up. You'll see the Docker whale icon in your system tray (Windows) or menu bar (macOS) when it's ready.
• Verify Docker is working by running this command in your terminal:

docker --version

You should see a version number printed, something like Docker version 28.x.x.

✔️ I see a version number

Docker is installed and running. You're ready to move on.

ⓧ Command not found

This means Docker Desktop didn't finish installing or isn't running yet.

• Make sure Docker Desktop is open and fully started (look for the whale icon in your system tray or menu bar).
• If you haven't installed it yet, go to https://www.docker.com/products/docker-desktop/ and download the installer for your operating system.
• After installing, restart your terminal and try docker --version again.

Install VS Code

VS Code is the code editor you'll use throughout this project to write Python scripts, Dockerfiles, and configuration files.

• Open your browser and go to the VS Code download page:

https://code.visualstudio.com/

• Download the installer for your operating system.
• Run the installer and follow the prompts. On Windows, the User setup is recommended (it does not require administrator permissions).
• Once installed, verify you can open VS Code by pressing Cmd+Space (macOS) or the Windows key (Windows) to open your search bar.
• Type Visual Studio Code and press Enter to open it.

VS Code not appearing in search?

If you just installed VS Code, you may need to restart your computer for it to appear in your system search. Alternatively, find the installer output folder and launch it directly.

help me open VS Code after installing it

Create the project folder and install dependencies

Your fraud detection pipeline has multiple components: a transaction simulator, streaming notebooks, training scripts, and an inference API. You'll organize these into separate subdirectories inside one project folder.

• Move to your Desktop by running this command:

cd ~/Desktop

• Create your project folder and move into it by running these commands:

mkdir fraud-detection
cd fraud-detection

• Create the subdirectories for each pipeline component by running:

mkdir simulator streaming training api

What are these folders for?

• simulator/ holds the Kafka producer that generates fake transactions.
• streaming/ stores exported Databricks notebook code for the medallion architecture.
• training/ contains the model training script and MLflow experiment logic.
• api/ holds the FastAPI inference endpoint and its Dockerfile.

• Create a Python virtual environment by running:

python -m venv venv

What is a virtual environment?

A virtual environment is an isolated Python installation inside your project folder. It keeps your project's packages separate from your system Python, preventing version conflicts between different projects.

• Activate the virtual environment by running:

macOS

source venv/bin/activate

Windows

venv\Scripts\activate

You should see (venv) appear at the beginning of your terminal prompt. This confirms the virtual environment is active.

• Create a requirements.txt file that lists all the Python packages your project needs. In VS Code, create a new file called requirements.txt in the fraud-detection folder and paste the following content:

# Kafka producer for transaction streaming
confluent-kafka

# Spark for streaming and batch processing
pyspark

# ML model training
xgboost
scikit-learn
numpy
pandas

# MLflow for experiment tracking (Azure ML requires <3.0)
mlflow<3.0

# FastAPI inference endpoint
fastapi
uvicorn

# Azure services
redis
azure-cosmos

# HTTP client for webhooks
httpx

Why pin mlflow to less than 3.0?

Azure Machine Learning does not yet support MLflow 3.x. Pinning to mlflow<3.0 ensures compatibility when you register your model in the Azure ML workspace later.

• Install all the dependencies by running:

pip install -r requirements.txt

What does this install?

This command reads every package listed in requirements.txt and installs them into your virtual environment. It may take a few minutes to download and compile everything.

Once the install finishes, you should see a success message in your terminal with no error output.

Seeing build errors during install?

Some packages (like confluent-kafka) use pre-built wheels that include all dependencies. If you see errors about missing C compilers or build tools, make sure you're running Python 3.11+ and that your pip is up to date.

• Upgrade pip by running: pip install --upgrade pip

help me fix Python package installation errors

• Confirm the project structure looks correct by running:

ls

You should see api, requirements.txt, simulator, streaming, training, and venv in the output.

✔️ Awesome, I've got everything!

Great. Make sure you've saved requirements.txt and that your virtual environment is still active (you should see (venv) in your terminal prompt).

ⓧ I'd like to double check the full code

Here's the complete requirements.txt file for comparison:

# Kafka producer for transaction streaming
confluent-kafka

# Spark for streaming and batch processing
pyspark

# ML model training
xgboost
scikit-learn
numpy
pandas

# MLflow for experiment tracking (Azure ML requires <3.0)
mlflow<3.0

# FastAPI inference endpoint
fastapi
uvicorn

# Azure services
redis
azure-cosmos

# HTTP client for webhooks
httpx

Your development environment is fully set up. Next up, you'll create your first Azure resources and start building the streaming pipeline.

Create the Azure Resource Group and Event Hubs Namespace

Your local environment is ready. Now it's time to create the cloud infrastructure that will receive and buffer the transaction stream.

In this step, you'll provision the first Azure resources for your pipeline. A resource group organizes all the services for this project. Inside it, you'll create an Event Hubs namespace with a Kafka-compatible endpoint so your Python producer can stream transactions using the industry-standard Kafka protocol.

In this step, get ready to:

• Create an Azure resource group to hold all pipeline resources.
• Create an Event Hubs namespace in Standard tier for Kafka compatibility.
• Retrieve and save the Kafka-compatible connection string.

Create the resource group

Azure resource groups act as containers that hold related resources for a solution. Putting everything in one group makes cleanup easy at the end of the project.

• Create the resource group by running this command:

az group create --name fraud-detection-rg --location eastus

What does this command do?

`--name fraud-detection-rg` gives the group a descriptive name you'll reference in every subsequent command.

`--location eastus` places the group in the East US region. All resources you create inside this group will default to this region.

• You should see JSON output with "provisioningState": "Succeeded" confirming the group was created.

Seeing an authorization error?

Make sure you're logged in to the correct Azure account. Run az account show to check which subscription is active.

If you have multiple subscriptions, set the correct one with az account set --subscription followed by your subscription ID.

Why is my az group create command failing?

Create the Event Hubs namespace and event hub

An Event Hubs namespace is a container that holds one or more event hubs. The Standard tier enables the Kafka-compatible endpoint on port 9093, which your transaction simulator will connect to in the next step.

Why Standard tier instead of Basic?

The Basic tier does not support the Kafka protocol. Standard tier enables the Kafka endpoint automatically, which means your Python producer can use the confluent-kafka library to send messages without any Event Hubs-specific SDK.

Standard tier costs approximately $0.03/hour per Throughput Unit. You'll only be charged for the time the namespace exists.

• Create the Event Hubs namespace by running this command:

az eventhubs namespace create \
  --name frauddetectionns \
  --resource-group fraud-detection-rg \
  --location eastus \
  --sku Standard

What does this command do?

`--name frauddetectionns` sets the namespace name. This must be globally unique across Azure (6-50 characters, letters, numbers, and hyphens only). If you get a conflict, add a few random digits to the end.

`--sku Standard` explicitly sets the Standard pricing tier. This enables the Kafka-compatible endpoint that your producer will use.

• Wait for the command to complete. This may take 1-2 minutes.
• In the JSON output, confirm you see "provisioningState": "Succeeded" and "kafkaEnabled": true.

Now create the event hub inside your namespace. This is the specific topic that your transaction simulator will publish messages to.

• Create the transactions event hub by running:

az eventhubs eventhub create \
  --name transactions \
  --resource-group fraud-detection-rg \
  --namespace-name frauddetectionns

What does this command do?

`--name transactions` creates an event hub named transactions inside your namespace. In Kafka terms, this is equivalent to creating a topic.

The event hub inherits the namespace's settings (Standard tier, East US region) and gets 4 partitions by default, which is plenty for this project's throughput.

• Confirm the output shows "status": "Active" in the JSON response.

Namespace name already taken?

Event Hubs namespace names are globally unique. If you see an error about the name being unavailable, append a few random digits (e.g., frauddetectionns4821). Use your updated name in all subsequent commands.

Help me fix my Event Hubs namespace creation error

Retrieve the connection string for Kafka access

Your transaction simulator needs a connection string to authenticate with Event Hubs via the Kafka protocol. Every namespace comes with a default authorization rule called RootManageSharedAccessKey that provides full access.

• Retrieve the connection string by running:

az eventhubs namespace authorization-rule keys list \
  --resource-group fraud-detection-rg \
  --namespace-name frauddetectionns \
  --name RootManageSharedAccessKey

What does this command do?

This fetches the primary and secondary connection strings for your namespace's default shared access policy.

The primaryConnectionString in the output is what your Kafka producer will use as the SASL password when connecting to port 9093.

• Copy the value of primaryConnectionString from the JSON output. It starts with Endpoint=sb://.
• In VS Code, create a new file called .env in your fraud-detection project folder.
• Add the following line, pasting your connection string as the value:

EVENT_HUBS_CONNECTION_STRING=your-connection-string-here

Why store this in a .env file?

Storing connection strings in environment files keeps secrets out of your source code. Your Python simulator will read this value at runtime using the os.environ module or a library like python-dotenv.

Never commit .env files to version control. If you use Git, add .env to your .gitignore.

• Save the .env file.
• Verify the file exists by running:

ls .env

• You should see .env listed in the output.

Connection string command returned empty?

Double-check that the --namespace-name matches exactly what you used when creating the namespace (including any random digits you added).

Run az eventhubs namespace list --resource-group fraud-detection-rg to see all namespaces in your resource group and confirm the name.

Help me retrieve my Event Hubs connection string

Your streaming infrastructure is live. You have a resource group, a Kafka-compatible Event Hubs namespace, and a transactions event hub ready to receive data. Next up, you'll build the Python simulator that generates realistic transactions and streams them into this event hub.

Build and Run the Transaction Simulator

Your Event Hubs namespace is live and waiting for data. Now it's time to build the component that generates that data.

Real fraud detection systems process thousands of credit card transactions per second. You'll build a Python simulator that creates realistic transaction records and streams them to Event Hubs using the Kafka protocol. This teaches you how production Kafka producers work, since the confluent-kafka client is the same one used at banks and fintechs.

In this step, get ready to:

• Write a Python Kafka producer that generates realistic credit card transactions.
• Configure SASL_SSL authentication to connect to Event Hubs on port 9093.
• Run the simulator and verify messages arrive in Azure Portal.

Write the transaction data generator

The simulator needs to produce realistic-looking transaction data. Each record will include a unique transaction ID, card ID, dollar amount, merchant name, category, location, and timestamp. About 1% of transactions will be flagged as fraudulent, mimicking real-world fraud rates.

• Install python-dotenv so your script can read the connection string from .env by running:

pip install python-dotenv

• Create the producer file by running:

touch simulator/producer.py

• Open simulator/producer.py in VS Code and paste the following code:

import json
import os
import random
import time
import uuid
from datetime import datetime, timezone

from confluent_kafka import Producer
from dotenv import load_dotenv

# Load environment variables from the project .env file
load_dotenv(os.path.join(os.path.dirname(__file__), "..", ".env"))

# Retrieve the Event Hubs connection string
connection_string = os.getenv("EVENT_HUBS_CONNECTION_STRING")

# Extract the namespace name from the connection string
namespace = connection_string.split("//")[1].split(".")[0]

# Kafka producer configuration for Azure Event Hubs
producer_config = {
    "bootstrap.servers": f"{namespace}.servicebus.windows.net:9093",
    "security.protocol": "SASL_SSL",
    "sasl.mechanism": "PLAIN",
    "sasl.username": "$ConnectionString",
    "sasl.password": connection_string,
    "client.id": "fraud-detection-simulator",
}

# Realistic sample data for transaction generation
MERCHANTS = [
    "Amazon", "Walmart", "Target", "Starbucks", "Shell Gas",
    "Netflix", "Uber", "DoorDash", "Apple Store", "Best Buy",
]

MERCHANT_CATEGORIES = [
    "online_retail", "grocery", "department_store", "food_beverage",
    "gas_station", "streaming", "rideshare", "food_delivery",
    "electronics", "electronics",
]

LOCATIONS = [
    "New York, NY", "Los Angeles, CA", "Chicago, IL", "Houston, TX",
    "Phoenix, AZ", "San Francisco, CA", "Seattle, WA", "Denver, CO",
    "Miami, FL", "Boston, MA",
]


def generate_transaction():
    """Generate a single realistic transaction record."""
    merchant_index = random.randint(0, len(MERCHANTS) - 1)

    # Approximately 1% of transactions are fraudulent
    is_fraud = random.random() < 0.01

    # Fraudulent transactions tend to have higher amounts
    if is_fraud:
        amount = round(random.uniform(500.0, 5000.0), 2)
    else:
        amount = round(random.uniform(1.0, 500.0), 2)

    transaction = {
        "transaction_id": str(uuid.uuid4()),
        "card_id": f"card_{random.randint(1000, 9999)}",
        "amount": amount,
        "merchant": MERCHANTS[merchant_index],
        "merchant_category": MERCHANT_CATEGORIES[merchant_index],
        "location": LOCATIONS[random.randint(0, len(LOCATIONS) - 1)],
        "timestamp": datetime.now(timezone.utc).isoformat(),
        "is_fraud": is_fraud,
    }

    return transaction

What does this code do?

The top section loads your Event Hubs connection string from .env and extracts the namespace name. It then builds the Kafka producer configuration that tells confluent-kafka how to connect to Event Hubs.

The generate_transaction() function creates a single transaction record with all the fields your fraud detection model will eventually need. The is_fraud flag is set to True about 1% of the time, and fraudulent transactions get higher dollar amounts (between $500 and $5000).

Why SASL_SSL on port 9093?

Azure Event Hubs exposes a Kafka-compatible endpoint on port 9093 (not the standard Kafka port 9092). Authentication uses the SASL_SSL protocol with the PLAIN mechanism.

The username is literally the string $ConnectionString and the password is your full Event Hubs connection string. This is specific to how Azure implements Kafka compatibility.

Add the producer loop that streams to Event Hubs

The generator function creates data, but you still need the loop that actually sends each transaction to Event Hubs. This is the producer's main event loop.

• In simulator/producer.py, add the following code below the generate_transaction() function:

def delivery_report(err, msg):
    """Callback for message delivery confirmation."""
    if err is not None:
        print(f"Delivery failed: {err}")
    else:
        print(f"Sent: {msg.topic()} [{msg.partition()}] @ {msg.offset()}")


def run_simulator():
    """Run the transaction simulator in a continuous loop."""
    producer = Producer(producer_config)
    topic = "transactions"

    print("Starting fraud detection transaction simulator...")
    print(f"Target topic: {topic}")
    print("Press Ctrl+C to stop.\n")

    count = 0
    try:
        while True:
            transaction = generate_transaction()

            # Serialize and send the transaction to Event Hubs
            producer.produce(
                topic=topic,
                value=json.dumps(transaction).encode("utf-8"),
                callback=delivery_report,
            )

            # Trigger delivery report callbacks
            producer.poll(0)

            count += 1
            if count % 10 == 0:
                print(f"Total transactions sent: {count}")

            # Random delay between transactions (0.5-2 seconds)
            time.sleep(random.uniform(0.5, 2.0))

    except KeyboardInterrupt:
        print(f"\nStopping simulator... {count} transactions sent.")
    finally:
        # Wait for any outstanding messages to be delivered
        producer.flush()
        print("Simulator stopped.")


if __name__ == "__main__":
    run_simulator()

What does this code do?

The delivery_report function is a callback that fires every time a message is successfully delivered (or fails). This gives you real-time visibility into whether your messages are reaching Event Hubs.

The run_simulator function creates the Kafka producer, generates transactions in an infinite loop, serializes each one as JSON, and sends it to the transactions topic. The producer.poll(0) call processes delivery callbacks without blocking.

When you press Ctrl+C, the KeyboardInterrupt is caught and producer.flush() ensures any buffered messages are sent before the script exits.

• Save simulator/producer.py.

✔️ Awesome, I've got everything!

Great. Double check you've saved the file before moving on.

ⓧ I'd like to double check the full code

import json
import os
import random
import time
import uuid
from datetime import datetime, timezone

from confluent_kafka import Producer
from dotenv import load_dotenv

# Load environment variables from the project .env file
load_dotenv(os.path.join(os.path.dirname(__file__), "..", ".env"))

# Retrieve the Event Hubs connection string
connection_string = os.getenv("EVENT_HUBS_CONNECTION_STRING")

# Extract the namespace name from the connection string
namespace = connection_string.split("//")[1].split(".")[0]

# Kafka producer configuration for Azure Event Hubs
producer_config = {
    "bootstrap.servers": f"{namespace}.servicebus.windows.net:9093",
    "security.protocol": "SASL_SSL",
    "sasl.mechanism": "PLAIN",
    "sasl.username": "$ConnectionString",
    "sasl.password": connection_string,
    "client.id": "fraud-detection-simulator",
}

# Realistic sample data for transaction generation
MERCHANTS = [
    "Amazon", "Walmart", "Target", "Starbucks", "Shell Gas",
    "Netflix", "Uber", "DoorDash", "Apple Store", "Best Buy",
]

MERCHANT_CATEGORIES = [
    "online_retail", "grocery", "department_store", "food_beverage",
    "gas_station", "streaming", "rideshare", "food_delivery",
    "electronics", "electronics",
]

LOCATIONS = [
    "New York, NY", "Los Angeles, CA", "Chicago, IL", "Houston, TX",
    "Phoenix, AZ", "San Francisco, CA", "Seattle, WA", "Denver, CO",
    "Miami, FL", "Boston, MA",
]


def generate_transaction():
    """Generate a single realistic transaction record."""
    merchant_index = random.randint(0, len(MERCHANTS) - 1)

    # Approximately 1% of transactions are fraudulent
    is_fraud = random.random() < 0.01

    # Fraudulent transactions tend to have higher amounts
    if is_fraud:
        amount = round(random.uniform(500.0, 5000.0), 2)
    else:
        amount = round(random.uniform(1.0, 500.0), 2)

    transaction = {
        "transaction_id": str(uuid.uuid4()),
        "card_id": f"card_{random.randint(1000, 9999)}",
        "amount": amount,
        "merchant": MERCHANTS[merchant_index],
        "merchant_category": MERCHANT_CATEGORIES[merchant_index],
        "location": LOCATIONS[random.randint(0, len(LOCATIONS) - 1)],
        "timestamp": datetime.now(timezone.utc).isoformat(),
        "is_fraud": is_fraud,
    }

    return transaction


def delivery_report(err, msg):
    """Callback for message delivery confirmation."""
    if err is not None:
        print(f"Delivery failed: {err}")
    else:
        print(f"Sent: {msg.topic()} [{msg.partition()}] @ {msg.offset()}")


def run_simulator():
    """Run the transaction simulator in a continuous loop."""
    producer = Producer(producer_config)
    topic = "transactions"

    print("Starting fraud detection transaction simulator...")
    print(f"Target topic: {topic}")
    print("Press Ctrl+C to stop.\n")

    count = 0
    try:
        while True:
            transaction = generate_transaction()

            # Serialize and send the transaction to Event Hubs
            producer.produce(
                topic=topic,
                value=json.dumps(transaction).encode("utf-8"),
                callback=delivery_report,
            )

            # Trigger delivery report callbacks
            producer.poll(0)

            count += 1
            if count % 10 == 0:
                print(f"Total transactions sent: {count}")

            # Random delay between transactions (0.5-2 seconds)
            time.sleep(random.uniform(0.5, 2.0))

    except KeyboardInterrupt:
        print(f"\nStopping simulator... {count} transactions sent.")
    finally:
        # Wait for any outstanding messages to be delivered
        producer.flush()
        print("Simulator stopped.")


if __name__ == "__main__":
    run_simulator()

Run the simulator and verify in Azure Portal

The code is ready. Now you'll run the simulator and confirm that transactions are actually arriving in your Event Hub.

• Start the simulator by running this command from your fraud-detection folder:

python simulator/producer.py

You should see delivery confirmations printing to your terminal, with a count every 10 messages.

• Let the simulator run for about 30 seconds so it sends at least 15-20 messages.
• Open the Azure Portal at https://portal.azure.com.
• Search for Event Hubs in the top search bar and select your namespace.
• Click on the transactions event hub in the list.
• In the Overview panel, look at the Messages chart. You should see incoming message counts climbing.
• Return to your terminal and press Ctrl+C to stop the simulator.

What should I see?

Your terminal output should look something like this:

Starting fraud detection transaction simulator...
Target topic: transactions
Press Ctrl+C to stop.

Sent: transactions [0] @ 0
Sent: transactions [0] @ 1
Sent: transactions [0] @ 2
...
Total transactions sent: 10
Sent: transactions [0] @ 10
...

In the Azure Portal, the Messages chart may take 1-2 minutes to refresh. If you don't see data immediately, wait a moment and refresh the page.

Seeing a delivery failure or connection error?

Check that your .env file contains the full connection string (it should start with Endpoint=sb://).

Make sure your Event Hubs namespace is Standard tier, not Basic. The Kafka endpoint is not supported on the Basic tier.

If you see a timeout error, check that your firewall or VPN is not blocking port 9093.

help me debug my Event Hubs Kafka connection error

Your transaction simulator is streaming live data into Event Hubs. Next up, you'll set up Azure Databricks to consume this stream and process it through the medallion architecture.

Set Up Azure Databricks Workspace

Your transactions are flowing into Event Hubs. Now you need a processing engine that can read those events in real time, transform them, and build features for your fraud detection model.

Azure Databricks provides managed Apache Spark clusters for large-scale data processing. In this step, you'll create a Databricks workspace, launch a single-node cluster, and install the library that connects Spark to Event Hubs.

In this step, get ready to:

• Create an Azure Databricks workspace inside your resource group.
• Launch a single-node cluster with the latest LTS Databricks Runtime.
• Install the Event Hubs Spark connector library on the cluster.

Create an Azure Databricks workspace via the Azure Portal

A Databricks workspace is the environment where you'll write notebooks, manage clusters, and run Spark jobs. You'll create one inside your existing resource group so all fraud-detection resources stay together.

• In the Azure Portal, click Create a resource in the top-left corner.
• Search for Azure Databricks and select it from the results.
• Click Create.
• Under Resource group, select fraud-detection-rg from the dropdown.
• In the Workspace name field, enter fraud-detection-dbx.
• For Region, select East US (matching your resource group).
• For Pricing Tier, select Premium.
• Leave Workspace type as Hybrid (classic).
• Click Review + Create, then click Create.

The deployment takes 2-5 minutes. Wait until you see a notification that says deployment is complete.

• Once the deployment completes, click Go to resource.
• On the workspace overview page, click Launch Workspace to open the Databricks UI in a new browser tab.

Why Premium tier?

Premium tier gives you access to role-based access control, audit logging, and advanced cluster features. For a portfolio project demonstrating production patterns, Premium shows you understand enterprise-grade configurations.

The cost difference is minimal for short-lived demo clusters. Your Azure free trial credit covers it.

Deployment stuck or failed?

If the deployment fails, check that your subscription isn't on the Free Trial plan. Databricks requires a pay-as-you-go subscription. You may also need to request a vCPU quota increase for your region.

Help me troubleshoot my Databricks workspace deployment failure in the Azure Portal.

Launch a single-node cluster

Clusters are the compute resources that run your Spark code. A single-node cluster uses one machine instead of multiple, which keeps costs low while still providing full Spark functionality for this demo workload.

• In the Databricks workspace sidebar, click Compute.
• Click the Create compute button.
• Enter a cluster name: fraud-detection-cluster.
• Under Performance, check the Single node checkbox.
• For Databricks Runtime Version, select the latest LTS version available (look for the one labeled LTS).
• Set Terminate after to 20 minutes of inactivity.
• Click Create.

The cluster takes 3-5 minutes to start. You'll see a green circle next to the cluster name when it's running.

Why a single-node cluster?

Databricks charges per DBU (Databricks Unit) based on VM size and number of nodes. A single-node cluster uses one Standard_DS3_v2 instance, costing roughly $1-2/hour. This is sufficient for our demo transaction volume.

The 20-minute auto-termination prevents burning credit when the cluster is idle between sessions.

Cluster won't start?

If you see a quota error, your subscription may not have enough vCPU allocation. Go to Subscription > Usage + quotas in the Azure Portal and request an increase for the Standard DSv2 family in East US.

Help me resolve a quota error when starting my Databricks cluster.

Install the Event Hubs Spark connector library

Your cluster needs the Event Hubs connector to read streaming data from your transactions event hub. This is a Maven library that you install directly on the cluster.

• On the Compute page, click your cluster name (fraud-detection-cluster).
• Click the Libraries tab.
• Click Install new.
• Select Maven as the library source.
• In the Coordinates field, enter:
• com.microsoft.azure:azure-eventhubs-spark_2.12:2.3.22
• In the Exclusions field, enter:
• net.minidev:json-smart
• Click Install.

Wait until the library status shows Installed with a green checkmark. This usually takes 1-2 minutes.

Why the exclusion?

The Event Hubs Spark connector has a dependency on json-smart with a version range that conflicts with newer Maven repositories. Adding the exclusion net.minidev:json-smart tells Databricks to skip that dependency. The cluster already includes a compatible version.

Library installation failed?

If you see an error mentioning unresolved dependencies, double-check that you entered net.minidev:json-smart in the Exclusions field. This resolves the most common installation failure for this library.

Also verify the coordinates are exactly com.microsoft.azure:azure-eventhubs-spark_2.12:2.3.22 with no extra spaces.

Help me fix my Event Hubs Spark connector installation failure on Databricks.

Your Databricks workspace is ready with a running cluster and the Event Hubs connector installed. Next up, you'll create a notebook that reads live transactions from Event Hubs using Spark Structured Streaming.

Build the Bronze Layer (Raw Ingestion with Spark Streaming)

Your Databricks cluster is connected to Event Hubs and ready to process streaming data. Now you need a pipeline that continuously reads incoming transactions and stores them for downstream processing.

In a production fraud detection system, raw data must land in a persistent store before any transformation happens. This is the Bronze layer of the medallion architecture, where you capture every message exactly as it arrives.

In this step, get ready to:

• Create a Databricks notebook that reads from Event Hubs using Spark Structured Streaming.
• Parse the raw JSON transaction data from the Event Hubs message body.
• Write the raw transactions to a Delta table in append mode with checkpointing.

Create the Bronze ingestion notebook

Databricks notebooks let you run PySpark code interactively on your cluster. You will create a notebook that connects to Event Hubs as a streaming source, reads raw transaction messages, and lands them in a Delta Lake table.

• In your Databricks workspace, click Workspace in the left sidebar.
• Click the Create button, then select Notebook.
• Name the notebook 01_bronze_ingestion.
• Set the language to Python and attach it to your running cluster.
• Click Create.

Your new notebook opens with an empty cell ready for code.

Set up the Event Hubs connection configuration by pasting the following code into the first cell:

# Cell 1: Event Hubs connection configuration
connection_string = "your-event-hubs-connection-string-here"

# Encrypt the connection string (required by the Event Hubs Spark connector)
ehConf = {
    "eventhubs.connectionString": sc._jvm.org.apache.spark.eventhubs.EventHubsUtils.encrypt(connection_string),
    "eventhubs.consumerGroup": "$Default"
}

What does this code do?

The connection_string variable holds your Event Hubs namespace connection string with the EntityPath=transactions suffix appended.

The EventHubsUtils.encrypt() call encrypts the connection string before passing it to Spark. This is required by the Event Hubs Spark connector (version 2.3.15 and above) to protect credentials in memory.

• Replace your-event-hubs-connection-string-here with the connection string you retrieved in the earlier step.
• Make sure your connection string ends with ;EntityPath=transactions. If it does not, append it manually.
• Run the cell by pressing Shift+Enter.

Seeing a Py4JNetworkError?

This error means the Event Hubs Spark connector library is not installed on your cluster. Return to the cluster's Libraries tab and confirm the com.microsoft.azure:azure-eventhubs-spark_2.12 Maven package is installed and the cluster has been restarted.

help me fix a Py4JNetworkError when using EventHubsUtils.encrypt in Databricks

Read from Event Hubs and parse the JSON body

Event Hubs delivers messages as binary in a column called body. You need to cast that binary to a string and then parse the JSON structure into individual columns matching your transaction schema.

• Create a new cell below and paste the following code to define the transaction schema and start reading from Event Hubs:

# Cell 2: Read streaming data from Event Hubs and parse JSON
from pyspark.sql.functions import col, from_json
from pyspark.sql.types import StructType, StructField, StringType, DoubleType, BooleanType, TimestampType

# Define the schema matching our transaction simulator output
transaction_schema = StructType([
    StructField("transaction_id", StringType(), False),
    StructField("card_id", StringType(), False),
    StructField("amount", DoubleType(), False),
    StructField("merchant", StringType(), False),
    StructField("merchant_category", StringType(), False),
    StructField("location", StringType(), False),
    StructField("timestamp", StringType(), False),
    StructField("is_fraud", BooleanType(), False)
])

# Read streaming data from Event Hubs
raw_stream = (
    spark.readStream
    .format("eventhubs")
    .options(**ehConf)
    .load()
)

# Parse the binary body column into structured fields
parsed_stream = (
    raw_stream
    .withColumn("body_string", col("body").cast("string"))
    .select(from_json(col("body_string"), transaction_schema).alias("transaction"))
    .select("transaction.*")
)

What does this code do?

The transaction_schema defines the expected shape of each JSON message: transaction_id, card_id, amount, merchant, merchant_category, location, timestamp, and is_fraud. This matches what your transaction simulator produces.

The spark.readStream.format("eventhubs") call opens a continuous streaming connection to your Event Hub. The from_json function parses the raw binary body into typed columns that downstream notebooks can work with.

• Run the cell by pressing Shift+Enter.

No output appears yet because you have not started writing the stream to a sink. The next cell handles that.

Write to the Bronze Delta table

The final piece connects your parsed stream to a Delta Lake table. Delta's transaction log guarantees exactly-once processing for streaming writes, so you will not get duplicate records even if the cluster restarts.

• Create a new cell and paste the following code to write the parsed stream to a Delta table:

# Cell 3: Write parsed transactions to the Bronze Delta table
bronze_query = (
    parsed_stream.writeStream
    .format("delta")
    .outputMode("append")
    .option("checkpointLocation", "dbfs:/checkpoints/bronze")
    .start("dbfs:/delta/bronze/transactions")
)

print("Bronze streaming query started. Writing to dbfs:/delta/bronze/transactions")

What does this code do?

The writeStream.format("delta") tells Spark to persist records as a Delta table. The outputMode("append") setting means only new rows are written each micro-batch.

The checkpointLocation at dbfs:/checkpoints/bronze stores the stream's progress. If the cluster restarts, Spark picks up exactly where it left off. The data itself lands at dbfs:/delta/bronze/transactions.

• Run the cell by pressing Shift+Enter.
• You should see the message Bronze streaming query started. Writing to dbfs:/delta/bronze/transactions printed below the cell.
• Wait about 30 seconds for the first micro-batch to process, then verify data is landing by creating a new cell and running:

# Cell 4: Verify Bronze table has data
df = spark.read.format("delta").load("dbfs:/delta/bronze/transactions")
print(f"Bronze table row count: {df.count()}")
df.show(5)

• Run the verification cell. You should see a row count greater than 0 and a table showing columns like transaction_id, card_id, amount, merchant, and is_fraud.

Seeing 0 rows in the Bronze table?

Make sure your transaction simulator is still running and sending messages to the transactions event hub. If the simulator stopped, restart it from your local terminal.

Also confirm the EntityPath=transactions is appended to your connection string. Without it, the connector does not know which event hub to read from.

help me debug why my Bronze Delta table has 0 rows when streaming from Event Hubs in Databricks

What is the medallion architecture?

The medallion architecture organizes your data lake into three quality tiers. Bronze holds raw, unvalidated data exactly as it arrives. Silver contains cleaned, validated, and enriched records. Gold stores aggregated, business-ready features optimized for analytics and ML.

Each layer builds on the previous one. If you discover a bug in your Silver transformations, you can always replay from Bronze without re-ingesting from the source.

✔️ Awesome, I've got everything!

Great. Make sure you have saved all cells in your 01_bronze_ingestion notebook and that the streaming query is still running (green circle indicator next to the cell).

ⓧ I'd like to double check the full code

Here is the complete notebook code across all cells:

# Cell 1: Event Hubs connection configuration
connection_string = "your-event-hubs-connection-string-here"

# Encrypt the connection string (required by the Event Hubs Spark connector)
ehConf = {
    "eventhubs.connectionString": sc._jvm.org.apache.spark.eventhubs.EventHubsUtils.encrypt(connection_string),
    "eventhubs.consumerGroup": "$Default"
}

# Cell 2: Read streaming data from Event Hubs and parse JSON
from pyspark.sql.functions import col, from_json
from pyspark.sql.types import StructType, StructField, StringType, DoubleType, BooleanType, TimestampType

# Define the schema matching our transaction simulator output
transaction_schema = StructType([
    StructField("transaction_id", StringType(), False),
    StructField("card_id", StringType(), False),
    StructField("amount", DoubleType(), False),
    StructField("merchant", StringType(), False),
    StructField("merchant_category", StringType(), False),
    StructField("location", StringType(), False),
    StructField("timestamp", StringType(), False),
    StructField("is_fraud", BooleanType(), False)
])

# Read streaming data from Event Hubs
raw_stream = (
    spark.readStream
    .format("eventhubs")
    .options(**ehConf)
    .load()
)

# Parse the binary body column into structured fields
parsed_stream = (
    raw_stream
    .withColumn("body_string", col("body").cast("string"))
    .select(from_json(col("body_string"), transaction_schema).alias("transaction"))
    .select("transaction.*")
)

# Cell 3: Write parsed transactions to the Bronze Delta table
bronze_query = (
    parsed_stream.writeStream
    .format("delta")
    .outputMode("append")
    .option("checkpointLocation", "dbfs:/checkpoints/bronze")
    .start("dbfs:/delta/bronze/transactions")
)

print("Bronze streaming query started. Writing to dbfs:/delta/bronze/transactions")

# Cell 4: Verify Bronze table has data
df = spark.read.format("delta").load("dbfs:/delta/bronze/transactions")
print(f"Bronze table row count: {df.count()}")
df.show(5)

Your Bronze layer is now continuously ingesting raw transactions from Event Hubs into a persistent Delta table. Next up, you will build the Silver and Gold layers to clean, validate, and engineer features from this raw data.

Build the Silver and Gold Layers (Feature Engineering)

Your Bronze layer is now capturing every raw transaction flowing through Event Hubs. Raw data alone is not enough for a fraud detection model. The model needs clean, validated records and computed features that reveal spending patterns over time.

In the medallion architecture, the Silver layer validates and cleans raw data, while the Gold layer computes business-ready aggregated features. You will build both layers in this step, creating the feature set your XGBoost model will train on.

In this step, get ready to:

• Create a Silver layer notebook that validates and type-casts Bronze transactions.
• Create a Gold layer notebook that computes per-card aggregated features for fraud detection.
• Write the Gold feature table to Delta format for model training.

Create the Silver layer notebook

The Silver layer acts as a quality gate. Transactions missing critical fields or carrying incorrect types would produce garbage features downstream.

• Return to your Azure Databricks workspace in the browser.
• Click + New in the left sidebar.
• Select Notebook from the menu.
• Click the notebook title at the top and rename it to 02_silver_processing.
• Make sure the notebook is attached to your single-node cluster (check the cluster dropdown at the top right of the notebook).
• Paste the following code into the first cell:

from pyspark.sql.functions import col
from pyspark.sql.types import DoubleType, TimestampType

# Read raw transactions from the Bronze Delta table
bronze_df = spark.readStream \
    .format("delta") \
    .load("dbfs:/delta/bronze/transactions")

# Validate data quality: drop rows missing critical fields
validated_df = bronze_df.filter(
    col("transaction_id").isNotNull() &
    col("card_id").isNotNull() &
    col("amount").isNotNull()
)

# Cast columns to correct types for downstream processing
silver_df = validated_df \
    .withColumn("amount", col("amount").cast(DoubleType())) \
    .withColumn("timestamp", col("timestamp").cast(TimestampType()))

# Write cleaned records to the Silver Delta table
silver_query = silver_df.writeStream \
    .format("delta") \
    .outputMode("append") \
    .option("checkpointLocation", "dbfs:/checkpoints/silver") \
    .start("dbfs:/delta/silver/transactions")

What does this code do?

This notebook reads from your Bronze Delta table as a streaming source. Each micro-batch of new transactions passes through two validation gates.

• The filter() call drops any rows where transaction_id, card_id, or amount is null. Incomplete records would corrupt downstream aggregations.
• The cast() calls enforce DoubleType for amount and TimestampType for timestamp. This ensures arithmetic and time-window operations work correctly in the Gold layer.
• The checkpoint at dbfs:/checkpoints/silver guarantees exactly-once processing. If the stream restarts, it picks up where it left off.

• Run the cell by pressing Shift+Enter.
• You should see a streaming query status bar appear below the cell showing the query is active and processing batches.

Stream not starting?

• Verify your Bronze ingestion notebook (01_bronze_ingestion) is still running. The Silver stream reads from the Bronze Delta table, so data must be flowing there first.
• Check that the path dbfs:/delta/bronze/transactions exists by running dbutils.fs.ls("dbfs:/delta/bronze/transactions") in a new cell.
• Make sure the notebook is attached to the same cluster as your Bronze notebook.

Still stuck? Help me debug why my Silver streaming query won't start in Databricks.

Create the Gold layer notebook

The Gold layer transforms clean transactions into the features your fraud detection model needs. Instead of looking at individual transactions in isolation, the model needs patterns: how fast is this card spending, what is its typical amount, how many different merchants does it use?

• Click + New in the left sidebar again.
• Select Notebook.
• Rename the notebook to 03_gold_features.
• Make sure it is attached to the same single-node cluster.
• Paste the following code into the first cell:

from pyspark.sql.functions import col, count, avg, max as spark_max, countDistinct, unix_timestamp, lag, expr, current_timestamp
from pyspark.sql.window import Window

# Read cleaned transactions from the Silver Delta table
silver_stream = spark.readStream \
    .format("delta") \
    .load("dbfs:/delta/silver/transactions")

def compute_gold_features(batch_df, batch_id):
    """Compute per-card_id aggregated features from the full Silver table."""
    # Read the complete Silver table for aggregation
    silver_full = spark.read.format("delta").load("dbfs:/delta/silver/transactions")

    # Define a window partitioned by card, ordered by time
    card_window = Window.partitionBy("card_id").orderBy("timestamp")

    # Compute transaction velocity: count of transactions in the last hour
    velocity_df = silver_full.groupBy("card_id").agg(
        count(
            expr("CASE WHEN timestamp > current_timestamp() - INTERVAL 1 HOUR THEN 1 END")
        ).alias("transaction_velocity_1h"),
        avg("amount").alias("avg_amount"),
        spark_max("amount").alias("max_amount"),
        countDistinct("merchant").alias("merchant_diversity")
    )

    # Compute time since last transaction per card
    with_lag = silver_full.withColumn(
        "prev_timestamp", lag("timestamp").over(card_window)
    )
    latest_per_card = with_lag.groupBy("card_id").agg(
        spark_max("timestamp").alias("latest_ts"),
        spark_max("prev_timestamp").alias("prev_ts")
    ).withColumn(
        "time_since_last_txn",
        unix_timestamp("latest_ts") - unix_timestamp("prev_ts")
    ).select("card_id", "time_since_last_txn")

    # Join all features into a single Gold table
    gold_df = velocity_df.join(latest_per_card, "card_id", "left")

    # Overwrite the Gold feature table with latest features
    gold_df.write.format("delta") \
        .mode("overwrite") \
        .save("dbfs:/delta/gold/card_features")

# Trigger feature computation on each new micro-batch from Silver
gold_query = silver_stream.writeStream \
    .foreachBatch(compute_gold_features) \
    .option("checkpointLocation", "dbfs:/checkpoints/gold") \
    .start()

What does this code do?

This notebook uses the foreachBatch pattern. Every time new records arrive in the Silver table, Spark triggers the compute_gold_features function to recompute the entire feature set.

• transaction_velocity_1h counts how many transactions each card made in the last hour. A burst of transactions is a strong fraud signal.
• avg_amount and max_amount capture spending magnitude. Fraudulent transactions often deviate sharply from a card's historical average.
• merchant_diversity counts distinct merchants per card. Fraudsters often test a stolen card across many merchants rapidly.
• time_since_last_txn measures seconds between the two most recent transactions. Very short gaps indicate automated fraud attempts.

The overwrite mode ensures the Gold table always reflects the latest feature snapshot. The model training step will read this snapshot directly.

Why foreachBatch instead of a direct streaming write?

Structured Streaming does not support aggregation functions like groupBy directly in append mode. The foreachBatch pattern lets you treat each micro-batch trigger as a signal to recompute Gold features using standard batch DataFrame operations against the full Silver table.

This is a common production pattern for feature stores. It decouples the streaming trigger from the aggregation logic.

• Run the cell by pressing Shift+Enter.
• You should see the streaming query status bar appear, confirming the Gold computation is active.

Seeing a AnalysisException error?

• If the error mentions the Silver path not existing, make sure the 02_silver_processing notebook is running and has processed at least one batch. The Gold notebook needs data in Silver before it can read.
• If you see a checkpoint conflict, delete the checkpoint folder by running dbutils.fs.rm("dbfs:/checkpoints/gold", recurse=True) and re-run the cell.

Still stuck? Help me debug my Gold layer foreachBatch streaming query in Databricks.

Verify the Gold feature table

Before moving on, confirm that all three layers are flowing and features are being computed correctly.

• Create a new cell in the 03_gold_features notebook by clicking the + button below the running cell.
• Query the Gold table by running this code:

# Preview the Gold feature table
gold_features = spark.read.format("delta").load("dbfs:/delta/gold/card_features")
gold_features.show(5)

• Press Shift+Enter to run the cell.
• You should see a table with columns card_id, transaction_velocity_1h, avg_amount, max_amount, merchant_diversity, and time_since_last_txn.

What should I see?

Each row represents one card's aggregated features. The transaction_velocity_1h column shows how many transactions that card made in the last hour. The time_since_last_txn column shows seconds between the two most recent transactions (it may be null for cards with only one transaction so far).

Table is empty?

• Make sure your transaction simulator is still running locally. The pipeline needs fresh data flowing through Event Hubs into Bronze and Silver before Gold features appear.
• Wait 30-60 seconds for the streaming pipeline to process at least one full cycle through all three layers, then re-run the query cell.

Still stuck? Help me debug why my Gold Delta table is empty in Databricks.

Your full medallion architecture is now streaming. Raw transactions flow into Bronze, get validated in Silver, and emerge as model-ready features in Gold. Next up, you will train an XGBoost fraud detection model using these features.

Train the Fraud Detection Model with MLflow

Your Gold layer now holds clean, aggregated features per card. The next challenge is turning those features into a model that can score transactions as fraudulent or legitimate in real time.

In this step, you will create an Azure Machine Learning workspace, train an XGBoost binary classifier on your Gold feature data, and register the trained model for later serving.

In this step, get ready to:

• Create an Azure Machine Learning workspace for experiment tracking.
• Train an XGBoost fraud detection model inside a Databricks notebook.
• Register the trained model in Azure ML via the MLflow API.

Create the Azure Machine Learning workspace

Azure ML workspaces are MLflow-compatible out of the box. This means you can use standard MLflow APIs from Databricks to log experiments directly into Azure ML without hosting your own tracking server.

• Install the Azure ML CLI extension by running this command in your local terminal:

az extension add -n ml

• Create the Azure ML workspace inside your existing resource group by running:

az ml workspace create -n fraud-detection-ml -g fraud-detection-rg -l eastus

What does this command do?

This creates an Azure ML workspace named fraud-detection-ml in your fraud-detection-rg resource group. Azure automatically provisions associated resources (Storage Account, Key Vault, Application Insights, Container Registry).

The workspace will take 2-3 minutes to provision. Wait for the command to complete before proceeding.

• Once the workspace is created, retrieve the MLflow tracking URI by running:

az ml workspace show --query mlflow_tracking_uri -g fraud-detection-rg -n fraud-detection-ml -o tsv

• Copy the output URI. It starts with azureml:// and you will paste it into your Databricks notebook shortly.

Seeing a permissions error?

Make sure you are logged in to Azure CLI with az login and that your subscription is set correctly. The free trial account should have Owner access to the resource group you created.

help me fix Azure ML workspace creation errors

Build the training notebook in Databricks

XGBoost is the industry standard for tabular fraud detection. It consistently matches or beats deep learning approaches on structured feature data while training orders of magnitude faster.

• In your Azure Databricks workspace, click Workspace in the left sidebar.
• Click the dropdown arrow next to your user folder and select Create, then Notebook.
• Name the notebook 04_model_training and set the language to Python.
• Attach the notebook to your running single-node cluster.

The first cell installs the required packages. Databricks notebooks use %pip to install notebook-scoped libraries.

• Add the following code in the first cell to install dependencies:

# Cell 1: Install required packages
%pip install xgboost scikit-learn "mlflow<3.0" azureml-mlflow

Why mlflow<3.0?

Azure Machine Learning does not yet support MLflow 3.x. Pinning to mlflow<3.0 ensures compatibility with the Azure ML model registry and tracking APIs.

The azureml-mlflow package provides the integration layer between MLflow and Azure ML for remote tracking.

• Run the cell. After installation completes, add a new cell below with this code to read the Gold feature dataset:

# Cell 2: Load the Gold features from Delta Lake
from pyspark.sql import SparkSession

spark = SparkSession.builder.getOrCreate()

# Read the aggregated features from the Gold layer
gold_df = spark.read.format("delta").load("dbfs:/delta/gold/card_features")

# Convert to pandas for single-node XGBoost training
df = gold_df.toPandas()
print(f"Dataset shape: {df.shape}")
print(df.head())

What does this code do?

This reads your Gold Delta Lake table containing the per-card aggregated features (transaction velocity, average amount, max amount, merchant diversity, and time since last transaction).

It converts the Spark DataFrame to pandas because XGBoost's single-node training operates on pandas or numpy arrays.

• Run Cell 2. You should see the dataset shape and first few rows printed in the output.
• Add a new cell to prepare features and split into train/test sets:

# Cell 3: Prepare features and split into train/test sets
import xgboost as xgb
from sklearn.model_selection import train_test_split
from sklearn.metrics import precision_score, recall_score, f1_score, roc_auc_score
import mlflow

# Define feature columns and target
feature_cols = ["transaction_velocity_1h", "avg_amount", "max_amount",
                "merchant_diversity", "time_since_last_txn"]
target_col = "is_fraud"

X = df[feature_cols]
y = df[target_col]

# Split into train/test sets (80/20)
X_train, X_test, y_train, y_test = train_test_split(
    X, y, test_size=0.2, random_state=42, stratify=y
)

# Calculate class imbalance ratio for scale_pos_weight
neg_count = (y_train == 0).sum()
pos_count = (y_train == 1).sum()
scale_pos_weight = neg_count / pos_count

print(f"Training set: {X_train.shape[0]} samples")
print(f"Test set: {X_test.shape[0]} samples")
print(f"Fraud ratio: {pos_count / len(y_train):.4f}")
print(f"scale_pos_weight: {scale_pos_weight:.2f}")

What does this code do?

The code separates your feature columns from the fraud label. It splits data 80/20 using stratified sampling to preserve the fraud ratio in both sets.

The scale_pos_weight calculation handles class imbalance. Fraud is rare (typically 1-2% of transactions), so the model needs to weight fraudulent examples more heavily during training. This ratio tells XGBoost how much more to penalize misclassifying fraud.

• Run Cell 3. You should see the training/test set sizes and the calculated fraud ratio printed.

Log the experiment to MLflow and register the model

Now you will configure MLflow to send tracking data to your Azure ML workspace. Then you will train the model inside an MLflow run that automatically logs parameters, metrics, and the model artifact.

• Add a new cell with the following code to configure the MLflow tracking URI:

# Cell 4: Configure MLflow tracking to Azure ML workspace
import mlflow

# Set the tracking URI to your Azure ML workspace
# Paste the URI you copied from the az ml workspace show command
mlflow_tracking_uri = "[[MLFLOW_TRACKING_URI=\"your azureml:// tracking URI from the CLI output\"]]"
mlflow.set_tracking_uri(mlflow_tracking_uri)

# Set experiment name
mlflow.set_experiment("fraud-detection-training")

print("MLflow tracking configured for Azure ML.")

What does this code do?

The set_tracking_uri() call points MLflow to your Azure ML workspace instead of the default Databricks-hosted tracking server. This ensures your experiment metrics and registered models appear in Azure ML Studio.

Replace the placeholder with the azureml:// URI you copied from the CLI output earlier.

• Run Cell 4. You should see the confirmation message printed.
• Add the final cell to train XGBoost, log everything to MLflow, and register the model:

# Cell 5: Train XGBoost and log to MLflow, then register the model
with mlflow.start_run(run_name="xgboost-fraud-v1") as run:
    # Define hyperparameters
    params = {
        "n_estimators": 100,
        "max_depth": 6,
        "learning_rate": 0.1,
        "scale_pos_weight": scale_pos_weight,
        "objective": "binary:logistic",
        "eval_metric": "aucpr",
        "random_state": 42
    }

    # Log parameters to MLflow
    mlflow.log_params(params)

    # Train the XGBoost classifier
    model = xgb.XGBClassifier(**params)
    model.fit(X_train, y_train)

    # Generate predictions on test set
    y_pred = model.predict(X_test)
    y_pred_proba = model.predict_proba(X_test)[:, 1]

    # Calculate evaluation metrics
    precision = precision_score(y_test, y_pred)
    recall = recall_score(y_test, y_pred)
    f1 = f1_score(y_test, y_pred)
    auc_roc = roc_auc_score(y_test, y_pred_proba)

    # Log metrics to MLflow
    mlflow.log_metric("precision", precision)
    mlflow.log_metric("recall", recall)
    mlflow.log_metric("f1_score", f1)
    mlflow.log_metric("auc_roc", auc_roc)

    # Log the model artifact
    mlflow.xgboost.log_model(model, artifact_path="model")

    print(f"Precision: {precision:.4f}")
    print(f"Recall: {recall:.4f}")
    print(f"F1 Score: {f1:.4f}")
    print(f"AUC-ROC: {auc_roc:.4f}")
    print(f"Run ID: {run.info.run_id}")

    # Register the model in Azure ML model registry
    model_uri = f"runs:/{run.info.run_id}/model"
    registered_model = mlflow.register_model(model_uri, "fraud-detection-xgboost")
    print(f"Model registered: {registered_model.name}, version {registered_model.version}")

What does this code do?

Inside the mlflow.start_run() context, every parameter, metric, and artifact is tracked together as a single experiment run.

The XGBoost classifier is configured with n_estimators (number of trees), max_depth (tree complexity), learning_rate (step size), and scale_pos_weight (class imbalance correction). After training, precision, recall, F1, and AUC-ROC are logged as metrics.

The mlflow.register_model() call takes the model artifact from this run and registers it in the Azure ML model registry under the name fraud-detection-xgboost. This makes it available for deployment.

• Run Cell 5. This trains the model and registers it. You should see the four metrics printed along with the run ID and model version number.

Seeing an authentication error when logging to Azure ML?

The Databricks cluster needs Azure credentials to communicate with Azure ML. Make sure your tracking URI matches the workspace you created and that you are logged in via az login on the machine that launched the cluster.

If you see a timeout, wait and re-run the cell. The first MLflow connection to Azure ML can take a moment to authenticate.

help me fix MLflow authentication errors with Azure ML from Databricks

• Verify the model registration by opening the Azure Portal, navigating to your fraud-detection-ml workspace, and clicking Models in the left menu.
• You should see fraud-detection-xgboost listed with Version 1.

✔️ Awesome, I've got everything!

Great. Double check you have saved all five cells in your 04_model_training notebook and that the model appears in Azure ML.

ⓧ I'd like to double check the full code

Here is the complete notebook code across all cells:

# Cell 1: Install required packages
%pip install xgboost scikit-learn "mlflow<3.0" azureml-mlflow

# Cell 2: Load the Gold features from Delta Lake
from pyspark.sql import SparkSession

spark = SparkSession.builder.getOrCreate()

# Read the aggregated features from the Gold layer
gold_df = spark.read.format("delta").load("dbfs:/delta/gold/card_features")

# Convert to pandas for single-node XGBoost training
df = gold_df.toPandas()
print(f"Dataset shape: {df.shape}")
print(df.head())

# Cell 3: Prepare features and split into train/test sets
import xgboost as xgb
from sklearn.model_selection import train_test_split
from sklearn.metrics import precision_score, recall_score, f1_score, roc_auc_score
import mlflow

# Define feature columns and target
feature_cols = ["transaction_velocity_1h", "avg_amount", "max_amount",
                "merchant_diversity", "time_since_last_txn"]
target_col = "is_fraud"

X = df[feature_cols]
y = df[target_col]

# Split into train/test sets (80/20)
X_train, X_test, y_train, y_test = train_test_split(
    X, y, test_size=0.2, random_state=42, stratify=y
)

# Calculate class imbalance ratio for scale_pos_weight
neg_count = (y_train == 0).sum()
pos_count = (y_train == 1).sum()
scale_pos_weight = neg_count / pos_count

print(f"Training set: {X_train.shape[0]} samples")
print(f"Test set: {X_test.shape[0]} samples")
print(f"Fraud ratio: {pos_count / len(y_train):.4f}")
print(f"scale_pos_weight: {scale_pos_weight:.2f}")

# Cell 4: Configure MLflow tracking to Azure ML workspace
import mlflow

# Set the tracking URI to your Azure ML workspace
# Paste the URI you copied from the az ml workspace show command
mlflow_tracking_uri = "[[MLFLOW_TRACKING_URI=\"your azureml:// tracking URI from the CLI output\"]]"
mlflow.set_tracking_uri(mlflow_tracking_uri)

# Set experiment name
mlflow.set_experiment("fraud-detection-training")

print("MLflow tracking configured for Azure ML.")

# Cell 5: Train XGBoost and log to MLflow, then register the model
with mlflow.start_run(run_name="xgboost-fraud-v1") as run:
    # Define hyperparameters
    params = {
        "n_estimators": 100,
        "max_depth": 6,
        "learning_rate": 0.1,
        "scale_pos_weight": scale_pos_weight,
        "objective": "binary:logistic",
        "eval_metric": "aucpr",
        "random_state": 42
    }

    # Log parameters to MLflow
    mlflow.log_params(params)

    # Train the XGBoost classifier
    model = xgb.XGBClassifier(**params)
    model.fit(X_train, y_train)

    # Generate predictions on test set
    y_pred = model.predict(X_test)
    y_pred_proba = model.predict_proba(X_test)[:, 1]

    # Calculate evaluation metrics
    precision = precision_score(y_test, y_pred)
    recall = recall_score(y_test, y_pred)
    f1 = f1_score(y_test, y_pred)
    auc_roc = roc_auc_score(y_test, y_pred_proba)

    # Log metrics to MLflow
    mlflow.log_metric("precision", precision)
    mlflow.log_metric("recall", recall)
    mlflow.log_metric("f1_score", f1)
    mlflow.log_metric("auc_roc", auc_roc)

    # Log the model artifact
    mlflow.xgboost.log_model(model, artifact_path="model")

    print(f"Precision: {precision:.4f}")
    print(f"Recall: {recall:.4f}")
    print(f"F1 Score: {f1:.4f}")
    print(f"AUC-ROC: {auc_roc:.4f}")
    print(f"Run ID: {run.info.run_id}")

    # Register the model in Azure ML model registry
    model_uri = f"runs:/{run.info.run_id}/model"
    registered_model = mlflow.register_model(model_uri, "fraud-detection-xgboost")
    print(f"Model registered: {registered_model.name}, version {registered_model.version}")

Your fraud detection model is trained, tracked, and registered. Next up, you will set up the Redis feature store and Cosmos DB so your inference API can look up features and store predictions in real time.

Set Up Redis Feature Store and Cosmos DB

Your fraud detection model is trained and registered, but it still needs a way to access features at inference time. When a new transaction arrives, the scoring API must look up that card's historical features instantly.

In this step, you will create two data stores that power the real-time inference layer. Azure Cache for Redis serves features with sub-millisecond latency. Azure Cosmos DB stores the prediction results for downstream analytics and dashboards.

In this step, get ready to:

• Create an Azure Cache for Redis instance (Basic C0 tier) for feature serving.
• Create an Azure Cosmos DB account with the free tier, plus a database and container for predictions.
• Write and run a Python script to materialize Gold features into Redis.

Create an Azure Cache for Redis instance

The inference API needs to look up a card's feature vector in under a millisecond. A Redis key-value store is ideal for this because every lookup is a simple hash retrieval by key.

• Create the Redis cache by running this command (this takes 10-20 minutes to provision):

az redis create --name fraud-detection-redis --resource-group fraud-detection-rg --location eastus --sku Basic --vm-size c0

What does this command do?

• Creates a 256 MB Redis cache on the Basic tier (single node, no SLA, ideal for development workloads).
• The --sku Basic and --vm-size c0 flags select the smallest (and cheapest) available size at approximately $0.022/hr.
• The cache is placed in fraud-detection-rg alongside your other project resources so a single az group delete cleans everything up later.

Redis takes time to provision

While Redis provisions, continue with the Cosmos DB commands below. You can run them in a second terminal tab or wait for Redis to finish first.

• Once the cache finishes provisioning, retrieve the hostname by running:

az redis show --name fraud-detection-redis --resource-group fraud-detection-rg --query hostName --output tsv

• Retrieve the primary access key by running:

az redis list-keys --name fraud-detection-redis --resource-group fraud-detection-rg --query primaryKey --output tsv

• Open your .env file in the fraud-detection folder and add these two lines, pasting in the values you just retrieved:

REDIS_HOST=your-redis-hostname-here
REDIS_KEY=your-redis-access-key-here

What is the hostname format?

The Redis hostname follows the pattern fraud-detection-redis.redis.cache.windows.net. This is the SSL endpoint your Python code will connect to on port 6380.

• Save the .env file.

Redis creation failed?

• The cache name must be globally unique, 1-63 characters, alphanumeric and hyphens only, starting and ending with a letter or number. If you get a naming conflict, try fraud-detection-redis-yourname.
• Make sure you are logged in with az login and your subscription has remaining credit.

Help me troubleshoot my Redis creation error.

Create an Azure Cosmos DB account, database, and container

Every fraud prediction your API makes needs to be stored somewhere for analytics, dashboarding, and auditing. Cosmos DB's free tier gives you 1000 RU/s and 25 GB of storage at no cost.

• Create a Cosmos DB account with the free tier enabled by running:

az cosmosdb create --name fraud-detection-cosmos --resource-group fraud-detection-rg --enable-free-tier true --default-consistency-level Session

Why Session consistency?

Session consistency guarantees that within a single client session, reads always see the latest writes from that session. This is the default for Cosmos DB and the best balance between performance and correctness for our prediction-writing use case.

• Create the fraud-detection-db database inside the Cosmos account by running:

az cosmosdb sql database create --account-name fraud-detection-cosmos --resource-group fraud-detection-rg --name fraud-detection-db

• Create the predictions container with a partition key of /card_id and 400 RU/s throughput by running:

az cosmosdb sql container create --account-name fraud-detection-cosmos --resource-group fraud-detection-rg --database-name fraud-detection-db --name predictions --partition-key-path /card_id --throughput 400

Why partition by card_id?

Partitioning by /card_id ensures all predictions for a given card are stored together. This makes querying fraud history for a specific card fast and efficient.

• Retrieve the Cosmos DB endpoint by running:

az cosmosdb show --name fraud-detection-cosmos --resource-group fraud-detection-rg --query documentEndpoint --output tsv

• Retrieve the Cosmos DB primary key by running:

az cosmosdb keys list --name fraud-detection-cosmos --resource-group fraud-detection-rg --query primaryMasterKey --output tsv

• Open your .env file again and add these two lines with the values you just retrieved:

COSMOS_ENDPOINT=your-cosmos-endpoint-here
COSMOS_KEY=your-cosmos-key-here

• Save the .env file.

Cosmos DB account name taken?

• Cosmos DB account names must be globally unique, lowercase, and fewer than 44 characters. Try appending your initials: fraud-detection-cosmos-abc.
• Only one free-tier Cosmos DB account is allowed per Azure subscription. If you already have one, either delete it or skip the --enable-free-tier true flag (standard billing will apply).

Help me fix my Cosmos DB creation error.

Materialize Gold features into Redis

The Gold Delta table contains the aggregated features your model was trained on. The inference API cannot query Databricks at runtime because the latency would be too high. Instead, you will copy those features into Redis where they can be retrieved in under a millisecond.

• Create the materialization script by running:

touch fraud-detection/scripts/materialize_features.py

• Confirm the file exists by running:

ls fraud-detection/scripts/

• You should see materialize_features.py in the output.
• Open fraud-detection/scripts/materialize_features.py in VS Code and paste the following code:

import os
import json
import redis
from dotenv import load_dotenv
from deltalake import DeltaTable

# Load environment variables from .env
load_dotenv()

# Connect to Azure Cache for Redis over SSL
redis_client = redis.StrictRedis(
    host=os.getenv("REDIS_HOST"),
    port=6380,
    password=os.getenv("REDIS_KEY"),
    ssl=True
)

# Read the Gold feature table from Delta Lake
print("Reading Gold features from Delta table...")
dt = DeltaTable("dbfs:/delta/gold/card_features")
df = dt.to_pandas()

print(f"Found {len(df)} card feature records.")

# Write each card's features to Redis as a JSON hash
for _, row in df.iterrows():
    card_id = row["card_id"]
    features = {
        "transaction_velocity_1h": float(row["transaction_velocity_1h"]),
        "avg_amount": float(row["avg_amount"]),
        "max_amount": float(row["max_amount"]),
        "merchant_diversity": float(row["merchant_diversity"]),
        "time_since_last_txn": float(row["time_since_last_txn"])
    }
    # Store with key pattern features:{card_id}
    redis_client.set(f"features:{card_id}", json.dumps(features))

print(f"Materialized {len(df)} feature records to Redis.")
print("Done!")

What does this script do?

• Connects to your Azure Redis instance over SSL on port 6380 using the credentials from .env.
• Reads the Gold Delta Lake table at dbfs:/delta/gold/card_features into a pandas DataFrame.
• Iterates through each card and writes a JSON object containing the five engineered features to Redis with the key pattern features:{card_id}.
• At inference time, the API simply calls redis_client.get(f"features:{card_id}") to retrieve everything it needs to score a transaction.

• Save materialize_features.py.
• Run the script to populate Redis by executing:

python fraud-detection/scripts/materialize_features.py

• You should see output similar to:

Reading Gold features from Delta table...
Found 150 card feature records.
Materialized 150 feature records to Redis.
Done!

Connection refused or SSL error?

• Double-check that REDIS_HOST in your .env matches the hostname from az redis show (should end in .redis.cache.windows.net).
• Make sure REDIS_KEY is the full primary key (a long base64 string). If you copied extra whitespace, the connection will fail silently.
• Verify the Redis instance has finished provisioning by checking its status in the Azure Portal under Azure Cache for Redis.

Help me fix my Redis connection error in the materialize script.

Your feature store is live and your prediction database is ready. Next up, you will build and deploy the FastAPI inference API that ties everything together.

Build and Deploy the FastAPI Inference API

Your model is trained, registered, and your Redis feature store is loaded with real-time features. The final piece of the inference pipeline is a live API that accepts incoming transactions, retrieves features, scores them against the model, and stores predictions.

In this step, you will build a FastAPI application that ties together Redis, MLflow, and Cosmos DB into a single inference endpoint. Then you will containerize it with Docker and deploy it to Azure Container Apps for live traffic.

In this step, get ready to:

• Write a FastAPI inference API that scores transactions against the fraud detection model.
• Create a Dockerfile to containerize the API.
• Deploy the container to Azure Container Apps with external ingress.

Write the FastAPI inference application

The inference API is the real-time decision layer. When a transaction arrives, it fetches the cardholder's historical features from Redis, combines them with the transaction amount, runs them through the XGBoost model, and writes the prediction to Cosmos DB.

• Create the api subdirectory inside your fraud-detection folder by running this command:

mkdir api

• Create a new file called main.py inside fraud-detection/api/ and add the following imports and application setup:

import os
import json
from datetime import datetime

import redis
import mlflow
import numpy as np
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from azure.cosmos import CosmosClient

# Initialize FastAPI app
app = FastAPI(title="Fraud Detection API")

# Redis connection for feature lookups
redis_client = redis.Redis(
    host=os.environ["REDIS_HOST"],
    port=6380,
    password=os.environ["REDIS_KEY"],
    ssl=True
)

# Cosmos DB connection for storing predictions
cosmos_client = CosmosClient(
    os.environ["COSMOS_ENDPOINT"],
    os.environ["COSMOS_KEY"]
)
database = cosmos_client.get_database_client("fraud-detection-db")
predictions_container = database.get_container_client("predictions")

# Load the MLflow model from Azure ML registry
mlflow.set_tracking_uri(os.environ["AZURE_ML_TRACKING_URI"])
model = mlflow.pyfunc.load_model("models:/fraud-detection-xgboost/1")


# Request schema for incoming transactions
class Transaction(BaseModel):
    transaction_id: str
    card_id: str
    amount: float
    merchant: str
    merchant_category: str
    location: str
    timestamp: str

What does this code do?

The top section sets up three external connections your API needs at startup. redis_client connects to Azure Cache for Redis over SSL on port 6380 to fetch pre-computed features. cosmos_client connects to your Cosmos DB predictions container to persist every scoring result.

The mlflow.pyfunc.load_model call downloads the fraud-detection-xgboost model Version 1 from the Azure ML registry at startup. This means the model loads once into memory, not on every request.

The Transaction Pydantic model defines the JSON schema that incoming requests must match. FastAPI automatically validates payloads against this schema.

• Now add the prediction endpoint below the Transaction class in the same file:

@app.post("/predict")
def predict(transaction: Transaction):
    # Look up pre-computed features from Redis
    feature_key = f"features:{transaction.card_id}"
    feature_data = redis_client.get(feature_key)

    if feature_data is None:
        raise HTTPException(
            status_code=404,
            detail=f"No features found for card {transaction.card_id}"
        )

    features = json.loads(feature_data)

    # Build feature array matching the model's training schema
    model_input = np.array([[
        transaction.amount,
        features["transaction_velocity_1h"],
        features["avg_amount"],
        features["max_amount"],
        features["merchant_diversity"],
        features["time_since_last_txn"]
    ]])

    # Score the transaction
    fraud_probability = float(model.predict(model_input)[0])
    is_fraud = fraud_probability > 0.5

    # Build prediction result
    prediction = {
        "id": transaction.transaction_id,
        "transaction_id": transaction.transaction_id,
        "card_id": transaction.card_id,
        "fraud_probability": fraud_probability,
        "is_fraud": is_fraud,
        "timestamp": datetime.utcnow().isoformat()
    }

    # Write prediction to Cosmos DB for analytics
    predictions_container.upsert_item(prediction)

    return prediction

What does the predict endpoint do?

The endpoint performs four steps in sequence. First, it looks up the cardholder's pre-computed features from Redis using the key pattern features:{card_id}. If no features exist for that card, it returns a 404.

Second, it builds a NumPy array combining the live transaction amount with the historical features (velocity, average amount, max amount, merchant diversity, time since last transaction). This matches the feature order the model was trained on.

Third, it scores the input through the XGBoost model and thresholds at 0.5 to produce a binary fraud decision. Finally, it writes the full prediction to Cosmos DB and returns it as JSON.

• Save api/main.py.

Seeing import errors in your editor?

Make sure your Python virtual environment is active. The packages (fastapi, redis, mlflow, azure-cosmos) are installed in the venv, not globally.

help me fix import errors in my FastAPI fraud detection app

• Next, create a requirements.txt file inside fraud-detection/api/ for the Docker image to install:

fastapi
uvicorn
redis
azure-cosmos
mlflow<3.0
azureml-mlflow
numpy
xgboost

Why pin mlflow<3.0?

Azure ML does not support MLflow 3.x yet. Pinning mlflow<3.0 ensures the container installs a compatible version that can communicate with the Azure ML tracking server and model registry.

• Save api/requirements.txt.

✔️ Awesome, I've got everything!

Great. Double check you have saved both main.py and requirements.txt inside fraud-detection/api/.

ⓧ I'd like to double check the full code

import os
import json
from datetime import datetime

import redis
import mlflow
import numpy as np
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from azure.cosmos import CosmosClient

# Initialize FastAPI app
app = FastAPI(title="Fraud Detection API")

# Redis connection for feature lookups
redis_client = redis.Redis(
    host=os.environ["REDIS_HOST"],
    port=6380,
    password=os.environ["REDIS_KEY"],
    ssl=True
)

# Cosmos DB connection for storing predictions
cosmos_client = CosmosClient(
    os.environ["COSMOS_ENDPOINT"],
    os.environ["COSMOS_KEY"]
)
database = cosmos_client.get_database_client("fraud-detection-db")
predictions_container = database.get_container_client("predictions")

# Load the MLflow model from Azure ML registry
mlflow.set_tracking_uri(os.environ["AZURE_ML_TRACKING_URI"])
model = mlflow.pyfunc.load_model("models:/fraud-detection-xgboost/1")


# Request schema for incoming transactions
class Transaction(BaseModel):
    transaction_id: str
    card_id: str
    amount: float
    merchant: str
    merchant_category: str
    location: str
    timestamp: str


@app.post("/predict")
def predict(transaction: Transaction):
    # Look up pre-computed features from Redis
    feature_key = f"features:{transaction.card_id}"
    feature_data = redis_client.get(feature_key)

    if feature_data is None:
        raise HTTPException(
            status_code=404,
            detail=f"No features found for card {transaction.card_id}"
        )

    features = json.loads(feature_data)

    # Build feature array matching the model's training schema
    model_input = np.array([[
        transaction.amount,
        features["transaction_velocity_1h"],
        features["avg_amount"],
        features["max_amount"],
        features["merchant_diversity"],
        features["time_since_last_txn"]
    ]])

    # Score the transaction
    fraud_probability = float(model.predict(model_input)[0])
    is_fraud = fraud_probability > 0.5

    # Build prediction result
    prediction = {
        "id": transaction.transaction_id,
        "transaction_id": transaction.transaction_id,
        "card_id": transaction.card_id,
        "fraud_probability": fraud_probability,
        "is_fraud": is_fraud,
        "timestamp": datetime.utcnow().isoformat()
    }

    # Write prediction to Cosmos DB for analytics
    predictions_container.upsert_item(prediction)

    return prediction

Create the Dockerfile

The Dockerfile packages your FastAPI app and all its dependencies into a container image that Azure Container Apps can run.

• Create a file called Dockerfile inside fraud-detection/api/ and add the following:

FROM python:3.11-slim

WORKDIR /app

# Install Python dependencies
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

# Copy application code
COPY . .

# Run the FastAPI app with Uvicorn
CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "8000"]

What does this Dockerfile do?

It starts from the official python:3.11-slim base image, which is a lightweight Linux image with Python pre-installed. The slim variant keeps the image small by excluding development tools you do not need at runtime.

The CMD instruction tells the container to start Uvicorn serving your FastAPI app on port 8000. This is the port that Azure Container Apps will route external traffic to.

• Save api/Dockerfile.
• Confirm all three files exist in the api/ folder by running:

ls api/

You should see Dockerfile, main.py, and requirements.txt listed.

Deploy to Azure Container Apps

Azure Container Apps provides a serverless container hosting service. The az containerapp up command handles everything in one step: it creates a container registry, builds your Docker image, pushes it, and deploys the container app with external ingress.

• Navigate into the api/ directory by running:

cd api

• Deploy the FastAPI app to Azure Container Apps by running this command:

az containerapp up --resource-group fraud-detection-rg --name fraud-detection-api --ingress external --target-port 8000 --source .

What does this command do?

The az containerapp up command with --source . performs several actions automatically. It creates an Azure Container Registry in your resource group, builds the Docker image from your Dockerfile, pushes the image to the registry, creates a Container Apps environment, and deploys the container app.

The --ingress external flag makes the API publicly accessible over HTTPS. The --target-port 8000 tells Container Apps to route traffic to port 8000 inside the container, which matches the Uvicorn port in your Dockerfile.

This command takes a few minutes

The build and deploy process typically takes 3-5 minutes. Watch the terminal output for the deployment URL that appears at the end.

• Once the deployment finishes, the CLI outputs the app's FQDN (fully qualified domain name). Copy this URL.
• Now set the environment variables your container needs. Run this command, replacing the placeholder values with your actual credentials from .env:

az containerapp update --name fraud-detection-api --resource-group fraud-detection-rg --set-env-vars REDIS_HOST=your-redis-host REDIS_KEY=your-redis-key COSMOS_ENDPOINT=your-cosmos-endpoint COSMOS_KEY=your-cosmos-key AZURE_ML_TRACKING_URI=your-tracking-uri

Why set environment variables separately?

The az containerapp up command deploys the image but does not accept secrets inline safely. Using az containerapp update --set-env-vars afterwards injects your Redis, Cosmos DB, and Azure ML credentials into the container runtime without embedding them in the image.

• Verify the deployment is live by sending a health check request to the API. Replace the URL with your Container App FQDN:

curl https://fraud-detection-api.<your-environment>.<region>.azurecontainerapps.io/docs

You should see the FastAPI automatic documentation page HTML returned. This confirms the container is running and accepting traffic.

• Save your Container App URL to your .env file as CONTAINER_APP_URL for use in the next step.

Deployment failed or container not starting?

Check that Docker Desktop is running. The az containerapp up command needs Docker to build the image locally before pushing.

If the container starts but crashes, check the logs with az containerapp logs show --name fraud-detection-api --resource-group fraud-detection-rg. Common issues include missing environment variables (the app tries to read REDIS_HOST at startup) or incorrect connection strings.

help me debug my Azure Container Apps deployment for the fraud detection API

Your inference API is live and ready to score transactions in real-time. Next up, you will run the full end-to-end pipeline test to verify everything works together.

Test the End-to-End Pipeline

You have built every component of the fraud detection system individually. The transaction simulator streams to Event Hubs, Databricks processes data through Bronze, Silver, and Gold layers, the model is registered in Azure ML, features are materialized in Redis, and the inference API is live on Container Apps.

Now it is time to prove the entire system works as one connected pipeline. You will send fresh transactions through the system and trace them all the way from ingestion to a fraud prediction stored in Cosmos DB.

In this step, get ready to:

• Run the transaction simulator and verify events arrive in Event Hubs.
• Confirm the streaming pipeline processes transactions through Bronze, Silver, and Gold layers.
• Test the inference API and verify predictions are stored in Cosmos DB.

Run the transaction simulator

The simulator generates realistic transaction data and streams it to Azure Event Hubs via the Kafka protocol. Running it now creates fresh data for the streaming pipeline to process.

• Make sure your virtual environment is active, then start the simulator by running:

python simulator/producer.py

• Let the simulator run for 30-60 seconds to generate a batch of transactions, then stop it with Ctrl+C.

What does this do?

The producer script generates transactions with random amounts, merchants, locations, and card IDs. Some transactions are flagged as fraudulent patterns (unusually high amounts, rapid sequences). Each transaction is serialized to JSON and published to the transactions event hub over port 9093 using SASL_SSL authentication.

• Verify messages arrived by opening the Azure Portal.
• Navigate to your Event Hubs namespace in the fraud-detection-rg resource group.
• On the Overview page, check the message counts chart. You should see a spike in Incoming Messages corresponding to the transactions you just sent.

Check metrics if the chart looks empty

Metrics can take a minute or two to appear. If the Overview chart still shows zero, select Monitoring then Metrics in the left menu. Choose Incoming Messages from the Metric dropdown and set the time range to Last 30 minutes.

No messages showing up at all?

Double-check that your EVENT_HUBS_CONNECTION_STRING in .env is correct and matches the namespace. The Kafka endpoint uses port 9093, not 9092.

Also confirm you are using the Standard tier namespace. The Kafka endpoint is not supported on the Basic tier.

Help me troubleshoot why my transaction simulator isn't sending messages to Event Hubs. I'm using confluent-kafka with SASL_SSL on port 9093.

Verify the streaming pipeline in Databricks

With fresh transactions in Event Hubs, the Spark Structured Streaming notebooks in Databricks should pick them up and process them through each layer of the medallion architecture.

• Switch to your Azure Databricks workspace.
• Open the 01_bronze_ingestion notebook and run it. This reads from Event Hubs and writes raw transactions to the Bronze Delta table.
• Open the 02_silver_processing notebook and run it. This validates and cleans the Bronze data into the Silver table.
• Open the 03_gold_features notebook and run it. This computes aggregated features per card ID and writes them to the Gold table.

• Verify data arrived in each layer by running a quick count query in a new notebook cell:

# Check record counts in each layer
bronze_count = spark.read.format("delta").load("dbfs:/delta/bronze/transactions").count()
silver_count = spark.read.format("delta").load("dbfs:/delta/silver/transactions").count()
gold_count = spark.read.format("delta").load("dbfs:/delta/gold/card_features").count()

print(f"Bronze records: {bronze_count}")
print(f"Silver records: {silver_count}")
print(f"Gold features: {gold_count}")

What should I see?

All three counts should be greater than zero. Bronze and Silver counts should be similar (Silver drops records that fail validation). Gold count represents the number of unique card IDs with computed features.

• Once Gold features are populated, switch back to your local terminal and materialize the features into Redis by running:

python scripts/materialize_features.py

• You should see output indicating features were written to Redis for each card ID found in the Gold table.

Seeing zero records in Bronze?

Ensure the 01_bronze_ingestion notebook has the correct Event Hubs connection string configured in its Spark connector settings. The streaming query needs a few seconds to start reading. Check that the cluster has the Event Hubs Spark connector library installed.

Help me debug why my Databricks streaming notebook isn't reading from Event Hubs.

Send a test request to the inference API

The final link in the pipeline is the FastAPI inference endpoint running on Azure Container Apps. It accepts a transaction, looks up features from Redis, scores it with the XGBoost model, and writes the prediction to Cosmos DB.

• Find your Container App URL. It is stored in your .env file as CONTAINER_APP_URL.
• Send a test transaction to the /predict endpoint by running this command (replace the URL with your own):

curl -X POST https://[[CONTAINER_APP_URL="your-container-app.azurecontainerapps.io"]]/predict \
  -H "Content-Type: application/json" \
  -d '{"card_id": "card_1001", "amount": 5000.00, "merchant": "Online Electronics Store", "location": "New York", "timestamp": "2026-06-02T14:30:00Z"}'

What does this request do?

The /predict endpoint receives the transaction JSON, fetches the card's historical features from Redis (keyed as features:card_1001), combines them with the transaction details, runs the XGBoost model to produce a fraud probability, and writes the result to Cosmos DB.

• You should see a JSON response similar to this:

{
  "transaction_id": "abc-123-def",
  "card_id": "card_1001",
  "fraud_probability": 0.87,
  "is_fraud": true,
  "timestamp": "2026-06-02T14:30:00Z"
}

What do these fields mean?

The fraud_probability is the model's confidence score between 0 and 1. The is_fraud field is true when the probability exceeds the decision threshold (typically 0.5). A high amount of $5,000 from a new merchant is likely to trigger a high fraud score with the model's learned patterns.

• Now verify the prediction was stored in Cosmos DB. Open the Azure Portal.
• Navigate to your Cosmos DB account in the fraud-detection-rg resource group.
• In the resource menu, select Data Explorer.
• Expand the fraud-detection-db database, then expand the predictions container, and select Items.
• You should see at least one document with your card_id, fraud_probability, and is_fraud fields matching the API response.

Getting a connection error from curl?

Make sure your Container App has external ingress enabled on port 8000. Check that the URL in your .env includes https:// and the full .azurecontainerapps.io domain.

If you get a 500 error, the API may not be able to reach Redis. Verify that REDIS_HOST and REDIS_KEY are correctly set as environment variables in the Container App configuration.

Help me debug why my FastAPI /predict endpoint is returning an error when I send a test request.

Your end-to-end pipeline is confirmed working. Transactions flow from the simulator through Event Hubs, get processed through the medallion architecture in Databricks, and the inference API scores them in real-time with predictions landing in Cosmos DB. Next up, you will tackle the Secret Mission and add a real-time alerting system on top of this pipeline.

Secret mission

Real-Time Fraud Alerting System

Your pipeline detects fraud, but nobody gets notified. In production, fraud detection without alerting is like a smoke detector without a siren. In this secret mission, you'll add a real-time alerting layer that flags high-confidence fraud, sends webhook notifications, rate-limits the inference endpoint, and provides a fraud summary dashboard query.

Clean Up Your Resources

Clean Up Your Resources

Decide whether to keep your resources running, pause them to come back later, or delete them entirely. This project uses several Azure services that incur hourly costs while active.

Cost warning

Your Event Hubs namespace costs ~$0.03/hr per Throughput Unit and Azure Cache for Redis costs ~$0.022/hr, even when idle. Your Databricks cluster costs ~$1-2/hr while running. If you leave everything up with the cluster stopped, expect ~$2-5/day in charges against your Azure free trial credit.

Cosmos DB and Container Apps are within their free tier limits for this project, so they do not add cost. If you are done with the project, deleting the resource group is the safest way to stop all charges immediately.

Resources you used:

• Azure Resource Group fraud-detection-rg (contains all cloud resources below).
• Event Hubs Namespace (Standard tier, Kafka-compatible).
• Azure Databricks Workspace with single-node cluster.
• Azure Cache for Redis (Basic C0 tier).
• Azure Cosmos DB Account (free tier).
• Azure Container Apps (FastAPI inference API).
• Azure Container Registry (Docker image storage).
• Azure ML Workspace (MLflow tracking and model registry).
• Local project folder fraud-detection/ on your Desktop.

Keep everything running

No action needed. Choose this if you are still actively building or want to demo your pipeline to others.

• Your Databricks cluster has auto-termination configured, so it stops after 20-30 minutes of inactivity. You only pay for the cluster while it is running.
• Event Hubs and Redis continue to incur charges (~$2-5/day total with the cluster stopped).
• Cosmos DB and Container Apps remain within free tier limits for demo-level traffic.
• Monitor your Azure free trial credit balance in the Azure Portal under Cost Management + Billing.

Pause - I'll come back to this later

Shut down running compute to reduce costs while keeping your resources intact. This drops your daily cost to ~$1-2/day (Event Hubs + Redis only).

• Stop the Databricks cluster. In the Azure Databricks workspace, go to Compute, find your cluster, and click Terminate.
• Scale Container Apps to zero replicas by running:

az containerapp update --name fraud-detection-api --resource-group fraud-detection-rg --min-replicas 0

• Your Delta tables, model registry, and Cosmos DB data all persist while paused.
• When you return, restart the Databricks cluster and scale Container Apps back up with --min-replicas 1.

Delete - I don't want to use this again

Remove all Azure resources and local files. This is the recommended option after completing the project. One command removes everything because all cloud resources live in the same resource group.

• Delete the entire Azure resource group by running:

az group delete --name fraud-detection-rg --yes --no-wait

What does this command do?

This deletes the resource group fraud-detection-rg and every resource inside it: Event Hubs, Databricks, Redis, Cosmos DB, Container Apps, Container Registry, and Azure ML. The --no-wait flag returns control to your terminal immediately while deletion continues in the background.

• Verify deletion has started by running:

az group list --output table

You should see fraud-detection-rg either missing from the list or showing a Deleting state.

• Delete the local project folder. On macOS/Linux:

cd ~/Desktop
rm -rf fraud-detection/

• On Windows (PowerShell):

cd ~\Desktop
Remove-Item -Recurse -Force fraud-detection\

• Remove local Docker images built for this project by running:

docker image prune -a

Keep Docker Desktop if you use containers

Docker Desktop is a general-purpose tool useful for many projects. Only uninstall it if you are certain you will not need containers again. You can uninstall it through your system's standard app removal flow (Settings > Apps on Windows, or drag to Trash on macOS).

• Log out of Azure CLI if you no longer need it:

az logout

Nice Work!

Nice Work!

You just built a production-grade real-time fraud detection system on Azure. From streaming raw transactions to scoring them with a trained ML model in milliseconds, you've implemented the full pipeline that financial institutions rely on every day.

Here's what you accomplished:

• Built a Kafka-compatible transaction simulator streaming to Azure Event Hubs, and processed transactions through a medallion architecture (Bronze, Silver, Gold) using Spark Structured Streaming on Azure Databricks.
• Trained an XGBoost fraud detection model with engineered time-window features, tracked experiments with MLflow, and registered the model in Azure ML.
• Deployed a FastAPI inference endpoint on Azure Container Apps with Redis for sub-millisecond feature lookups and Cosmos DB for storing predictions.
• Built a real-time alerting system with fraud notification webhooks and rate limiting on the inference endpoint.

Ready to quiz yourself?