# mohammadsaif001/elt-lakehouse architecture

> Architecture diagram of the GitHub repository mohammadsaif001/elt-lakehouse, generated by GitDiagram from its file tree, README and a sample of source files (last updated 2026-08-13). It is an AI-made overview: check details against the source.

- Interactive diagram: https://gitdiagram.com/mohammadsaif001/elt-lakehouse
- Repository: https://github.com/mohammadsaif001/elt-lakehouse

## Overview

#### Purpose and Entry Points
This repository implements a synthetic e-commerce lakehouse: generate Olist-shaped entities, stream/load raw records, transform them through Bronze/Silver/Gold layers, and publish warehouse analytics. Primary executable/orchestration entry points are `generators/build_dataset.py`, `ingestion/bronze/run.py`, `ingestion/kafka/kafka-to-bronze.py`, Spark jobs under `spark/jobs/`, and the Airflow DAG `airflow/dags/ecommerce_pipeline.py`.

#### 1. Synthetic Source Dataset and Referential Pools
Generation is dependency-aware: seller/customer/product pools support orders; orders and items support payments and reviews. Profiling distributions in `data/profiling/` drive realistic values, while persistent pools establish cross-entity foreign-key consistency. Outputs are JSON datasets in `storage/generated/`.
- `generators/build_dataset.py`
- `generators/base/pool_manager.py`
- `metadata/pools/customer_pool.json`

#### 2. Event and Raw-Ingestion Boundary
Generated/source records are moved through Kafka or batch bootstrap into the Bronze/raw layer. Kafka producer/consumer/topic management are isolated from generic ingestion abstractions, allowing Bronze ingestion to be the storage boundary rather than downstream transformation logic.
- `ingestion/kafka/producer.py`
- `ingestion/kafka/consumer.py`
- `ingestion/core/ingestor.py`

Kafka configuration is centralized in `config/kafka_config.yaml`; bootstrap loaders exist for initial CSV-derived source loading under `ingestion/bootstrap/`.

#### 3. Bronze Lake and Spark Processing
Bronze jobs persist raw inputs, then Spark Silver jobs clean and standardize entity datasets. Shared Spark session, schema registry, and paths form the runtime contract between jobs. Incremental support includes CDC, merge handling, and watermarks/checkpoints, so repeated processing can maintain state rather than requiring full rebuilds.
- `spark/jobs/bronze_job.py`
- `spark/jobs/silver_job.py`
- `spark/incremental/merge_handler.py`

State artifacts are represented in `metadata/checkpoints/bronze_checkpoint.json` and `metadata/watermarks/watermark.json`; Silver storage is rooted at `storage/silver/`.

#### 4. Quality and Schema Enforcement
Quality checks are a first-class stage around Spark processing, covering schema validation, nulls, duplicates, freshness, and referential integrity. Entity contracts define expected input structures; configurable rules supply policy. Failures can be surfaced through Airflow callbacks/alerts.
- `spark/jobs/quality_job.py`
- `spark/quality/schema_validation.py`
- `contracts/order_schema.json`

Relevant control/configuration boundaries are `config/dq_rules.yaml`, `airflow/include/callbacks.py`, and `airflow/include/alerts.py`.

#### 5. Analytics Modeling and Warehouse Publication
dbt models convert staged operational data into intermediate metrics, dimensional marts, and Gold analytics such as revenue, customer segmentation, payment analysis, and seller scorecards. PostgreSQL loader/schema scripts provide the serving warehouse boundary.
- `dbt/models/staging/stg_orders.sql`
- `dbt/models/marts/fact_orders.sql`
- `dbt/models/gold/monthly_revenue.sql`

Warehouse publishing is implemented through `warehouse/loaders/gold_to_postgres.py`, with relational schema managed in `warehouse/postgres/create_tables.sql`.

#### 6. Orchestration, Runtime, and Operations
Airflow coordinates pipeline stages and invokes custom dbt/Spark operators. Docker Compose supplies the local multi-service runtime: Airflow, Spark, Kafka, and PostgreSQL. Prometheus collects metrics and Grafana visualizes Kafka, Spark, and pipeline health; GitHub workflows run CI and specialized dbt/Spark validation.
- `airflow/dags/ecommerce_pipeline.py`
- `airflow/plugins/spark_operator.py`
- `docker/docker-compose.yml`

Operational assets include `monitoring/prometheus/prometheus.yml`, `monitoring/grafana/dashboards/pipeline_dashboard.json`, and `.github/workflows/ci.yml`.

Main entry points: [`airflow/dags/ecommerce_pipeline.py`](https://github.com/mohammadsaif001/elt-lakehouse/blob/main/airflow/dags/ecommerce_pipeline.py) (Pipeline DAG), [`spark/jobs/bronze_job.py`](https://github.com/mohammadsaif001/elt-lakehouse/blob/main/spark/jobs/bronze_job.py) (Bronze Spark job), [`ingestion/kafka/producer.py`](https://github.com/mohammadsaif001/elt-lakehouse/blob/main/ingestion/kafka/producer.py) (Kafka event boundary), [`spark/jobs/silver_job.py`](https://github.com/mohammadsaif001/elt-lakehouse/blob/main/spark/jobs/silver_job.py) (Silver Spark job)

## Diagram (Mermaid)

```mermaid
flowchart TD

subgraph group_source["Source generation"]
  node_dataset_builder["Dataset builder<br/>Python entry point<br/>[build_dataset.py]"]
  node_referential_pools[("Referential pools<br/>Persistent entity state<br/>[pool_manager.py]")]
  node_generated_data[("Generated JSON datasets<br/>Source data")]
end

subgraph group_ingestion["Ingestion and lakehouse"]
  node_kafka["Kafka event boundary<br/>Message broker<br/>[producer.py]"]
  node_ingestor["Bronze ingestor<br/>Ingestion abstraction<br/>[ingestor.py]"]
  node_bronze_job["Bronze Spark job<br/>[bronze_job.py]"]
  node_bronze_state["Checkpoints and watermarks<br/>Incremental state"]
  node_silver_job["Silver Spark job<br/>[silver_job.py]"]
  node_merge_handler["CDC merge handler<br/>Incremental processing<br/>[merge_handler.py]"]
  node_silver_lake[("Silver lake<br/>Curated storage")]
  node_quality_job["Quality job<br/>Spark job<br/>[quality_job.py]"]
  node_order_contract["Order schema contract<br/>Data contract<br/>[order_schema.json]"]
end

subgraph group_analytics["Analytics serving"]
  node_dbt_staging["dbt staging<br/>dbt model<br/>[stg_orders.sql]"]
  node_dbt_marts["Dimensional marts<br/>dbt model<br/>[fact_orders.sql]"]
  node_gold_analytics["Gold analytics<br/>dbt model"]
  node_warehouse_loader["PostgreSQL publisher<br/>Warehouse loader"]
  node_postgres[("Analytics warehouse<br/>PostgreSQL<br/>[create_tables.sql]")]
end

subgraph group_operations["Orchestration and operations"]
  node_airflow["Pipeline DAG<br/>Airflow orchestration"]
  node_runtime{{"Local service runtime<br/>Docker Compose<br/>[docker-compose.yml]"}}
  node_monitoring["Pipeline monitoring<br/>Prometheus and Grafana<br/>[prometheus.yml]"]
end

node_referential_pools -->|"supplies consistent IDs"| node_dataset_builder
node_dataset_builder -->|"writes entities"| node_generated_data
node_generated_data -->|"streams records"| node_kafka
node_generated_data -.->|"batch bootstrap"| node_ingestor
node_kafka -->|"consumes events"| node_ingestor
node_ingestor -->|"raw inputs"| node_bronze_job
node_bronze_job -->|"updates state"| node_bronze_state
node_bronze_job -->|"raw records"| node_silver_job
node_bronze_state -->|"incremental state"| node_merge_handler
node_merge_handler -->|"applies CDC merges"| node_silver_job
node_silver_job -->|"writes curated entities"| node_silver_lake
node_silver_lake -->|"validates datasets"| node_quality_job
node_order_contract -->|"expected schema"| node_quality_job
node_silver_lake -->|"models source data"| node_dbt_staging
node_dbt_staging -->|"builds facts and dimensions"| node_dbt_marts
node_dbt_marts -->|"derives business metrics"| node_gold_analytics
node_gold_analytics -->|"publishes gold outputs"| node_warehouse_loader
node_warehouse_loader -->|"loads serving tables"| node_postgres
node_airflow -.->|"orchestrates"| node_bronze_job
node_airflow -.->|"orchestrates"| node_silver_job
node_airflow -.->|"orchestrates and alerts"| node_quality_job
node_airflow -.->|"orchestrates dbt"| node_dbt_staging
node_runtime -.->|"runs"| node_airflow
node_runtime -.->|"runs"| node_kafka
node_runtime -.->|"runs"| node_postgres
node_monitoring -.->|"observes pipeline health"| node_airflow
node_monitoring -.->|"observes broker health"| node_kafka

click node_dataset_builder "https://github.com/mohammadsaif001/elt-lakehouse/blob/main/generators/build_dataset.py"
click node_referential_pools "https://github.com/mohammadsaif001/elt-lakehouse/blob/main/generators/base/pool_manager.py"
click node_kafka "https://github.com/mohammadsaif001/elt-lakehouse/blob/main/ingestion/kafka/producer.py"
click node_ingestor "https://github.com/mohammadsaif001/elt-lakehouse/blob/main/ingestion/core/ingestor.py"
click node_bronze_job "https://github.com/mohammadsaif001/elt-lakehouse/blob/main/spark/jobs/bronze_job.py"
click node_bronze_state "https://github.com/mohammadsaif001/elt-lakehouse/blob/main/metadata/checkpoints/bronze_checkpoint.json"
click node_silver_job "https://github.com/mohammadsaif001/elt-lakehouse/blob/main/spark/jobs/silver_job.py"
click node_merge_handler "https://github.com/mohammadsaif001/elt-lakehouse/blob/main/spark/incremental/merge_handler.py"
click node_quality_job "https://github.com/mohammadsaif001/elt-lakehouse/blob/main/spark/jobs/quality_job.py"
click node_order_contract "https://github.com/mohammadsaif001/elt-lakehouse/blob/main/contracts/order_schema.json"
click node_dbt_staging "https://github.com/mohammadsaif001/elt-lakehouse/blob/main/dbt/models/staging/stg_orders.sql"
click node_dbt_marts "https://github.com/mohammadsaif001/elt-lakehouse/blob/main/dbt/models/marts/fact_orders.sql"
click node_gold_analytics "https://github.com/mohammadsaif001/elt-lakehouse/blob/main/dbt/models/gold/monthly_revenue.sql"
click node_warehouse_loader "https://github.com/mohammadsaif001/elt-lakehouse/blob/main/warehouse/loaders/gold_to_postgres.py"
click node_postgres "https://github.com/mohammadsaif001/elt-lakehouse/blob/main/warehouse/postgres/create_tables.sql"
click node_airflow "https://github.com/mohammadsaif001/elt-lakehouse/blob/main/airflow/dags/ecommerce_pipeline.py"
click node_runtime "https://github.com/mohammadsaif001/elt-lakehouse/blob/main/docker/docker-compose.yml"
click node_monitoring "https://github.com/mohammadsaif001/elt-lakehouse/blob/main/monitoring/prometheus/prometheus.yml"

classDef toneNeutral fill:#f8fafc,stroke:#334155,stroke-width:1.5px,color:#0f172a
classDef toneBlue fill:#dbeafe,stroke:#2563eb,stroke-width:1.5px,color:#172554
classDef toneAmber fill:#fef3c7,stroke:#d97706,stroke-width:1.5px,color:#78350f
classDef toneMint fill:#dcfce7,stroke:#16a34a,stroke-width:1.5px,color:#14532d
classDef toneRose fill:#ffe4e6,stroke:#e11d48,stroke-width:1.5px,color:#881337
classDef toneIndigo fill:#e0e7ff,stroke:#4f46e5,stroke-width:1.5px,color:#312e81
classDef toneTeal fill:#ccfbf1,stroke:#0f766e,stroke-width:1.5px,color:#134e4a
class node_dataset_builder,node_referential_pools,node_generated_data toneBlue
class node_kafka,node_ingestor,node_bronze_job,node_bronze_state,node_silver_job,node_merge_handler,node_silver_lake,node_quality_job,node_order_contract toneAmber
class node_dbt_staging,node_dbt_marts,node_gold_analytics,node_warehouse_loader,node_postgres toneMint
class node_airflow,node_runtime,node_monitoring toneRose
```

## Components

### Source generation

- [Dataset builder](https://github.com/mohammadsaif001/elt-lakehouse/blob/main/generators/build_dataset.py): Python entry point · Builds dependency-aware synthetic e-commerce entities. · `generators/build_dataset.py`
- [Referential pools](https://github.com/mohammadsaif001/elt-lakehouse/blob/main/generators/base/pool_manager.py): Persistent entity state · Preserves cross-entity foreign-key consistency. · `generators/base/pool_manager.py`
- **Generated JSON datasets**: Source data

### Ingestion and lakehouse

- [Kafka event boundary](https://github.com/mohammadsaif001/elt-lakehouse/blob/main/ingestion/kafka/producer.py): Message broker · `ingestion/kafka/producer.py`
- [Bronze ingestor](https://github.com/mohammadsaif001/elt-lakehouse/blob/main/ingestion/core/ingestor.py): Ingestion abstraction · Defines the raw-storage ingestion boundary. · `ingestion/core/ingestor.py`
- [Bronze Spark job](https://github.com/mohammadsaif001/elt-lakehouse/blob/main/spark/jobs/bronze_job.py): Spark job · `spark/jobs/bronze_job.py`
- [Checkpoints and watermarks](https://github.com/mohammadsaif001/elt-lakehouse/blob/main/metadata/checkpoints/bronze_checkpoint.json): Incremental state · Maintains repeatable CDC and incremental processing state. · `metadata/checkpoints/bronze_checkpoint.json`
- [Silver Spark job](https://github.com/mohammadsaif001/elt-lakehouse/blob/main/spark/jobs/silver_job.py): Spark job · Cleans and standardizes entity datasets. · `spark/jobs/silver_job.py`
- [CDC merge handler](https://github.com/mohammadsaif001/elt-lakehouse/blob/main/spark/incremental/merge_handler.py): Incremental processing · `spark/incremental/merge_handler.py`
- **Silver lake**: Curated storage
- [Quality job](https://github.com/mohammadsaif001/elt-lakehouse/blob/main/spark/jobs/quality_job.py): Spark job · Checks schema, completeness, freshness, and integrity. · `spark/jobs/quality_job.py`
- [Order schema contract](https://github.com/mohammadsaif001/elt-lakehouse/blob/main/contracts/order_schema.json): Data contract · `contracts/order_schema.json`

### Analytics serving

- [dbt staging](https://github.com/mohammadsaif001/elt-lakehouse/blob/main/dbt/models/staging/stg_orders.sql): dbt model · `dbt/models/staging/stg_orders.sql`
- [Dimensional marts](https://github.com/mohammadsaif001/elt-lakehouse/blob/main/dbt/models/marts/fact_orders.sql): dbt model · `dbt/models/marts/fact_orders.sql`
- [Gold analytics](https://github.com/mohammadsaif001/elt-lakehouse/blob/main/dbt/models/gold/monthly_revenue.sql): dbt model · Revenue and other business analytics. · `dbt/models/gold/monthly_revenue.sql`
- [PostgreSQL publisher](https://github.com/mohammadsaif001/elt-lakehouse/blob/main/warehouse/loaders/gold_to_postgres.py): Warehouse loader · `warehouse/loaders/gold_to_postgres.py`
- [Analytics warehouse](https://github.com/mohammadsaif001/elt-lakehouse/blob/main/warehouse/postgres/create_tables.sql): PostgreSQL · `warehouse/postgres/create_tables.sql`

### Orchestration and operations

- [Pipeline DAG](https://github.com/mohammadsaif001/elt-lakehouse/blob/main/airflow/dags/ecommerce_pipeline.py): Airflow orchestration · Coordinates ingestion, Spark, quality, and dbt stages. · `airflow/dags/ecommerce_pipeline.py`
- [Local service runtime](https://github.com/mohammadsaif001/elt-lakehouse/blob/main/docker/docker-compose.yml): Docker Compose · Runs Airflow, Spark, Kafka, and PostgreSQL locally. · `docker/docker-compose.yml`
- [Pipeline monitoring](https://github.com/mohammadsaif001/elt-lakehouse/blob/main/monitoring/prometheus/prometheus.yml): Prometheus and Grafana · Tracks Kafka, Spark, and pipeline health. · `monitoring/prometheus/prometheus.yml`

## Connections

- Referential pools → Dataset builder: supplies consistent IDs
- Dataset builder → Generated JSON datasets: writes entities
- Generated JSON datasets → Kafka event boundary: streams records
- Generated JSON datasets → Bronze ingestor: batch bootstrap
- Kafka event boundary → Bronze ingestor: consumes events
- Bronze ingestor → Bronze Spark job: raw inputs
- Bronze Spark job → Checkpoints and watermarks: updates state
- Bronze Spark job → Silver Spark job: raw records
- Checkpoints and watermarks → CDC merge handler: incremental state
- CDC merge handler → Silver Spark job: applies CDC merges
- Silver Spark job → Silver lake: writes curated entities
- Silver lake → Quality job: validates datasets
- Order schema contract → Quality job: expected schema
- Silver lake → dbt staging: models source data
- dbt staging → Dimensional marts: builds facts and dimensions
- Dimensional marts → Gold analytics: derives business metrics
- Gold analytics → PostgreSQL publisher: publishes gold outputs
- PostgreSQL publisher → Analytics warehouse: loads serving tables
- Pipeline DAG → Bronze Spark job: orchestrates
- Pipeline DAG → Silver Spark job: orchestrates
- Pipeline DAG → Quality job: orchestrates and alerts
- Pipeline DAG → dbt staging: orchestrates dbt
- Local service runtime → Pipeline DAG: runs
- Local service runtime → Kafka event boundary: runs
- Local service runtime → Analytics warehouse: runs
- Pipeline monitoring → Pipeline DAG: observes pipeline health
- Pipeline monitoring → Kafka event boundary: observes broker health