🏦 Data Engineering Pipeline Platform

Enterprise-grade financial data platform for ingestion, validation, transformation, lineage tracking, and analytics delivery.

Overview • Features • Tech Stack • Screenshots • Dashboard • Architecture • Quick Start • API • Troubleshooting

---

📌 Overview

Data Engineering Pipeline Platform is a production-oriented data platform that ingests financial-style customer, account, transaction, and payment data; validates data quality; transforms datasets through bronze, silver, and gold layers; records lineage; and exposes operational metadata through FastAPI.

The project is designed to resemble a modern financial-institution data platform with orchestration, streaming, batch processing, warehouse modeling, observability, CI/CD, containerization, Kubernetes manifests, and AWS-ready deployment documentation.

---

💼 Business Problem

Financial organizations depend on trusted data pipelines for reporting, fraud analytics, customer intelligence, payment operations, and regulatory workflows.

When data quality fails, downstream dashboards, risk models, reconciliation jobs, and executive reports become unreliable. This platform demonstrates how enterprise data teams build auditable, monitored, and analytics-ready pipelines using modern data engineering practices.

---

✨ Features

🔄 Data Pipelines Batch ETL runner Airflow DAG orchestration Kafka producer and consumer PySpark transformation job Bronze / silver / gold zones dbt staging and mart models

✅ Data Reliability Schema validation Required-column checks Primary-key uniqueness checks Amount and balance quality rules Great Expectations-ready structure Failed-record metadata tracking

🚀 Platform Engineering FastAPI metadata service PostgreSQL warehouse schema Docker Compose support Kubernetes deployment manifests GitHub Actions CI Jenkins pipeline

---

🧱 Tech Stack

FastAPI Metadata API	Airflow Orchestration	Kafka Streaming	PySpark Batch Processing
PostgreSQL Warehouse	dbt Analytics Engineering	Great Expectations Validation	Docker Containerization
Kubernetes Deployment	AWS Cloud Target	GitHub Actions CI/CD	Jenkins Pipeline

---

📸 Screenshots

Dashboard

Swagger Endpoints

Airflow DAG

---

📊 Operational Dashboard

The platform includes a lightweight FastAPI-powered dashboard for local demos and portfolio walkthroughs. It gives a browser-based view of pipeline execution, quality checks, lineage events, and warehouse tables without requiring a separate frontend framework.

http://localhost:8000/dashboard

The dashboard includes:

• Pipeline run metrics and recent executions
• Data quality result summaries
• Source-to-target lineage events
• Warehouse table inventory
• Quick links to Swagger and Airflow

---

🏗️ Architecture

flowchart LR
    A[Financial Source Systems] --> B[Batch CSV Files]
    A --> C[Kafka Transaction Events]

    B --> D[Airflow DAG]
    C --> E[Kafka Consumer]

    D --> F[Bronze Raw Zone]
    E --> F

    F --> G[Data Quality Validation]
    G --> H[Silver Validated Zone]

    H --> I[PySpark Transformations]
    I --> J[Gold Analytics Zone]

    J --> K[dbt Models]
    K --> L[(PostgreSQL Warehouse)]

    L --> M[FastAPI Metadata API]
    M --> N[Pipeline Runs]
    M --> O[Quality Results]
    M --> P[Lineage Events]
    M --> Q[Warehouse Tables]

Loading

🔄 End-to-End Workflow

Financial Systems Produce Customer, Account, Transaction, and Payment Data
        ↓
Batch Files and Kafka Events Enter the Platform
        ↓
Airflow Orchestrates Batch Workflows
        ↓
Airflow Injects DAG Run ID Into the Batch Pipeline
        ↓
Batch Pipeline Writes Operational Metadata to FastAPI/PostgreSQL Tables
        ↓
Kafka Consumer Handles Streaming Transaction Events
        ↓
Raw Data Lands in the Bronze Layer
        ↓
Validation Rules and Quality Checks Execute
        ↓
Validated Records Move to the Silver Layer
        ↓
PySpark Jobs Transform and Aggregate Data
        ↓
Gold Layer Produces Curated Analytics Outputs
        ↓
dbt Builds Warehouse Models and Reporting Marts
        ↓
PostgreSQL Stores Warehouse and Metadata Tables
        ↓
FastAPI Exposes Pipeline, Quality, Lineage, and Warehouse Status

System Flow

Step	What Happens
1	Source systems provide customer, account, transaction, and payment data
2	Batch ingestion and Kafka streaming bring data into the platform
3	Bronze raw datasets are created for traceability
4	Data quality rules validate schema, required fields, uniqueness, and numeric constraints
5	Silver datasets contain cleaned and validated records
6	PySpark transformations generate analytical aggregates
7	dbt models create reporting-ready marts
8	FastAPI exposes operational metadata for observability

---

📁 Folder Structure

data-engineering-pipeline-platform/
├── app/                         # FastAPI metadata API
│   ├── api/                     # API routes
│   ├── core/                    # Configuration and security
│   ├── database/                # SQLAlchemy session
│   ├── models/                  # ORM models
│   ├── schemas/                 # Pydantic schemas
│   └── services/                # Business logic
├── airflow/
│   └── dags/                    # Airflow orchestration DAGs
├── kafka/
│   ├── consumers/               # Streaming consumers
│   ├── producers/               # Streaming producers
│   └── schemas/                 # Event schemas
├── spark/
│   └── jobs/                    # PySpark transformation jobs
├── dbt/
│   └── models/                  # dbt staging and mart models
├── great_expectations/          # Data quality project structure
├── database/                    # Warehouse schema and seed scripts
├── data/
│   ├── sample/                  # Sample financial datasets
│   ├── bronze/                  # Raw landed outputs
│   ├── silver/                  # Validated outputs
│   └── gold/                    # Curated analytics outputs
├── kubernetes/                  # Kubernetes manifests
├── monitoring/                  # Monitoring assets
├── pipelines/                   # Pipeline configuration
├── scripts/                     # Local runnable pipeline scripts
├── tests/                       # Unit and integration tests
├── Dockerfile
├── docker-compose.yml
├── Jenkinsfile
├── pyproject.toml
├── requirements.txt
└── README.md

---

⚡ Quick Start

Prerequisites

Requirement	Version
Python	3.11+
Docker	Recommended
Git	Any recent version
PostgreSQL	Optional for local metadata DB
Java	Required for full PySpark runtime

Create Environment

python -m venv .venv
source .venv/bin/activate
pip install -r requirements.txt

For Windows Git Bash:

python -m venv .venv
source .venv/Scripts/activate
pip install -r requirements.txt

Run API Locally

uvicorn app.main:app --reload

Open:

Swagger API: http://localhost:8000/docs
Dashboard:   http://localhost:8000/dashboard

Run Batch Pipeline

python scripts/run_batch_pipeline.py

Run Tests

pytest

Run Quality and Formatting Checks

ruff check .
black --check .
pytest

Run with Docker

docker compose up --build

Open:

Swagger API: http://localhost:8000/docs
Dashboard:   http://localhost:8000/dashboard
Airflow UI:  http://localhost:8080/home

---

🔌 API Reference

Health Check

curl http://localhost:8000/health

Operational Dashboard

http://localhost:8000/dashboard

List Pipelines

curl http://localhost:8000/pipelines \
  -H "x-api-key: dev-api-key"

Trigger Pipeline

curl -X POST http://localhost:8000/pipelines/daily_financial_batch/trigger \
  -H "x-api-key: dev-api-key"

List Pipeline Runs

curl http://localhost:8000/pipelines/daily_financial_batch/runs \
  -H "x-api-key: dev-api-key"

Read Pipeline Run

curl http://localhost:8000/pipeline-runs/{run_id} \
  -H "x-api-key: dev-api-key"

Read Quality Results

curl http://localhost:8000/quality/results \
  -H "x-api-key: dev-api-key"

Read Lineage Events

curl http://localhost:8000/lineage \
  -H "x-api-key: dev-api-key"

Read Warehouse Tables

curl http://localhost:8000/warehouse/tables \
  -H "x-api-key: dev-api-key"

---

🛡️ Data Platform Guardrails

✅ Included API key protection Data quality checks Pipeline run tracking Lineage event capture Warehouse schema design Local + Docker execution Kubernetes manifests CI/CD workflows

🔮 Enterprise Extensions OpenLineage integration Marquez lineage UI Prometheus metrics Grafana dashboards Terraform infrastructure Helm charts Spark-on-Kubernetes jobs Managed AWS deployment

---

🧪 What This Project Demonstrates

Skill Area	Demonstrated Through
Data Engineering	Batch ETL, ELT modeling, bronze/silver/gold architecture
Backend Engineering	FastAPI metadata APIs, Pydantic schemas, service layer
Streaming Systems	Kafka producer and consumer structure
Big Data Processing	PySpark transformation job
Analytics Engineering	dbt staging and mart models
Data Quality	Great Expectations-ready structure and validation rules
Platform Engineering	Docker Compose and Kubernetes manifests
DevOps	GitHub Actions and Jenkins pipeline
Cloud Architecture	AWS EKS, S3, MSK, RDS, CloudWatch strategy
Production Thinking	Observability, lineage, testing, security, and troubleshooting

---

☁️ AWS Deployment Strategy

S3
├── raw/
├── bronze/
├── silver/
└── gold/

MWAA / Airflow
└── DAG orchestration

MSK
└── Streaming transaction events

EKS
├── FastAPI metadata service
├── Kafka consumers
├── Spark jobs
└── supporting workers

RDS PostgreSQL
├── warehouse tables
├── pipeline run metadata
├── quality result metadata
└── lineage metadata

CloudWatch
├── application logs
├── pipeline metrics
└── operational alerts

Secrets Manager + IAM
└── credential and access management

---

🗄️ Database Separation

The Docker environment separates application metadata from Airflow orchestration metadata:

Database	Purpose	Used By
data_platform	Warehouse tables, pipeline runs, quality results, lineage events	FastAPI, batch pipeline, dbt-style outputs
airflow_metadata	DAG runs, task instances, scheduler state, Airflow users	Apache Airflow

This avoids mixing application/warehouse metadata with Airflow's internal operational tables and more closely resembles production data-platform deployments.

The dedicated Airflow database is created by:

database/init-airflow-db.sql

Airflow-to-Dashboard Metadata Flow

Airflow executions now feed the dashboard directly. The DAG passes its Airflow run ID into the batch script with PIPELINE_RUN_ID, while the Airflow container uses the application metadata database through DATABASE_URL.

Airflow DAG Run
    ↓
PIPELINE_RUN_ID='{{ run_id }}'
    ↓
scripts/run_batch_pipeline.py
    ↓
pipeline_runs / data_quality_results / lineage_events
    ↓
FastAPI /dashboard

This keeps Airflow as the orchestrator while the FastAPI dashboard becomes the operational observability layer for application-level pipeline metadata.

---

🔐 Security Considerations

Application Security API key authentication Environment-based configuration No committed secrets .env.example for local setup Pydantic request/response validation

Platform Security IAM least privilege design Kubernetes Secrets support Network boundary awareness Dependency scanning support Bandit-ready Python security checks

---

📊 Monitoring Strategy

Layer	Monitoring Focus
Airflow	DAG status, retries, duration
Kafka	Consumer lag, topic throughput
PySpark	Job status, execution time, failure rate
Great Expectations	Validation pass/fail status
PostgreSQL	Warehouse table freshness and growth
FastAPI	Request latency, error rate, health status
CI/CD	Build, test, lint, and security scan status

---

🧰 Troubleshooting

Ruff E402: Module level import not at top of file

This usually happens when path setup code appears before imports.

Run:

ruff check . --fix

If needed, keep path setup before project imports and mark project-level imports:

from app.database.session import SessionLocal  # noqa: E402

ModuleNotFoundError for app imports

Run commands from the repository root.

For macOS/Linux:

export PYTHONPATH=.

For Windows PowerShell:

$env:PYTHONPATH="."

For Windows Git Bash:

export PYTHONPATH=.

API key authentication fails in Swagger

Use this value in the x-api-key field:

dev-api-key

Or update the value in your local environment configuration.

PostgreSQL connection refused

Check whether services are running:

docker compose ps

Restart containers:

docker compose up --build

Kafka service unavailable

Start the Kafka service through Docker Compose:

docker compose up kafka

If the project uses a separate broker dependency, start the full stack:

docker compose up --build

PySpark job fails locally

Confirm Java is installed:

java -version

Then verify PySpark dependencies:

python -c "import pyspark; print(pyspark.__version__)"

Swagger docs load but endpoints return empty data

Run the batch pipeline first:

python scripts/run_batch_pipeline.py

Then refresh:

http://localhost:8000/docs

---

🔄 Recommended Clean Rebuild

rm -rf .pytest_cache .ruff_cache __pycache__
rm -f local_metadata.db
python -m venv .venv
source .venv/bin/activate
pip install -r requirements.txt
ruff check .
black --check .
pytest
python scripts/run_batch_pipeline.py
uvicorn app.main:app --reload

For Windows Git Bash, activate with:

source .venv/Scripts/activate

---

🗺️ Roadmap

Priority	Improvement
High	Add OpenLineage and Marquez integration
High	Add Prometheus /metrics endpoint
Medium	Add Grafana dashboard JSON
Medium	Add Terraform infrastructure for AWS
Medium	Add Helm charts
Low	Add Spark-on-Kubernetes job operator manifests
Low	Add dbt docs publishing workflow
Low	Add S3-backed local development mode

---

📄 License

This project is licensed under the MIT License.