How anaasih/moroccan-fmcg-data-pipeline works
This is a Moroccan FMCG retail ETL and data-warehouse pipeline. Its main workflow is scheduled by Airflow: ingest sales into raw storage with batch/file tracking, seed store and product dimensions, validate and transform sales into accepted facts or quarantined rejects, then expose warehouse data through seven BI marts. The graph preserves the documented warehouse and reporting capabilities. Source excerpts establish the Python calls and database operations; mart-to-dashboard relationships come from the README, since the SQL and reporting implementations were not sampled.
Main entry points: transform_sales.py Sales Quality Transform · pipeline.py ETL Pipeline · raw_ingestion.py Sales Ingestion · seed_dimensions.py Dimension Seeding
Components 17
Source Data
- Raw Source Files
data/raw
Orchestration
- Airflow DAG
dags/moroccan_retail_dag.py
ETL Processing
- ETL Pipeline
src/retail_etl/pipeline.py - Sales Ingestion
src/retail_etl/extract/raw_ingestion.py - Dimension Seeding
src/retail_etl/transform/seed_dimensions.py - Sales Quality Transform
src/retail_etl/transform/transform_sales.py - Source Profiling
src/retail_etl/profiling.py
Data Warehouse
- Database Connection
src/retail_etl/database/connection.py - Raw Sales
sql/ddl/01_create_schemas_and_tables.sql - Batch and File Tracking
sql/ddl/01_create_schemas_and_tables.sql - Store Dimension
sql/ddl/01_create_schemas_and_tables.sql - Product Dimension
sql/ddl/01_create_schemas_and_tables.sql - Sales Fact
sql/ddl/01_create_schemas_and_tables.sql - Rejected Records
sql/ddl/01_create_schemas_and_tables.sql
Analytics
- Seven BI Marts
sql/marts/02_create_bi_views.sql - Power BI Dashboard
Other components
- Daily Scheduler
Connections 24
- Daily Scheduler → Airflow DAG: schedules daily
README.md - Airflow DAG → ETL Pipeline: calls pipeline
dags/moroccan_retail_dag.py - ETL Pipeline → Sales Ingestion: calls ingestion
src/retail_etl/pipeline.py - ETL Pipeline → Dimension Seeding: seeds dimensions
src/retail_etl/pipeline.py - ETL Pipeline → Sales Quality Transform: runs transform
src/retail_etl/pipeline.py - ETL Pipeline → Database Connection: queries summary
src/retail_etl/pipeline.py - Raw Source Files → Sales Ingestion: provides sales CSV
src/retail_etl/extract/raw_ingestion.py - Sales Ingestion → Database Connection: gets engine
src/retail_etl/extract/raw_ingestion.py - Sales Ingestion → Raw Sales: loads raw records
src/retail_etl/extract/raw_ingestion.py - Sales Ingestion → Batch and File Tracking: records batch manifest
src/retail_etl/extract/raw_ingestion.py - Raw Source Files → Dimension Seeding: provides stores products
src/retail_etl/transform/seed_dimensions.py - Dimension Seeding → Database Connection: gets engine
src/retail_etl/transform/seed_dimensions.py - Dimension Seeding → Store Dimension: seeds stores
src/retail_etl/transform/seed_dimensions.py - Dimension Seeding → Product Dimension: seeds products
src/retail_etl/transform/seed_dimensions.py - Sales Quality Transform → Database Connection: gets engine
src/retail_etl/transform/transform_sales.py - Sales Quality Transform → Raw Sales: reads batch rows
src/retail_etl/transform/transform_sales.py - Sales Quality Transform → Store Dimension: checks store IDs
src/retail_etl/transform/transform_sales.py - Sales Quality Transform → Product Dimension: checks product IDs
src/retail_etl/transform/transform_sales.py - Sales Quality Transform → Sales Fact: writes accepted rows
src/retail_etl/transform/transform_sales.py - Sales Quality Transform → Rejected Records: quarantines invalid rows
src/retail_etl/transform/transform_sales.py - Sales Fact → Seven BI Marts: feeds sales views
README.md - Rejected Records → Seven BI Marts: feeds quality views
README.md - Seven BI Marts → Power BI Dashboard: powers reporting
README.md - Raw Source Files → Source Profiling: profiles source files
src/retail_etl/profiling.py