FMCG Sales Data Integration Pipeline

Project Overview

This project simulates a real-world post-acquisition data integration scenario. A parent FMCG company ("Atlon") acquires a sports-nutrition brand ("SportsBar") and needs its sales data merged into the parent's existing OLAP star schema for unified BI reporting.

Built using Databricks, PySpark, Delta Lake, AWS S3, and Databricks Lakeflow job, the pipeline ingests raw CSV files from Amazon S3, processes them through the Bronze → Silver → Gold Medallion Architecture, and merges the transformed data into the parent company's star schema using Delta Lake MERGE operations.

The solution supports both a one-time historical backfill and ongoing incremental data processing, producing an analytics-ready reporting layer for business intelligence and decision-making.

Business Context

Following Atlon's acquisition of Sports Bar, the organization faced critical data integration challenges that prevented unified business intelligence and strategic decision-making. The parent company (Atlon) operated a mature OLAP infrastructure with established BI capabilities, while the acquired entity (Sports Bar) relied on fragmented spreadsheets and ad-hoc reporting systems.

Business Problem

Challenge	Impact
Incompatible data formats between parent and acquired systems	No unified reporting across entities
Acquired company relied on spreadsheets, no reliable pipeline	Manual, error-prone, non-scalable reporting
5 months of historical sales data missing from analytics platform	Blind spots in trend and performance analysis
No incremental processing strategy	Reprocessing full datasets was inefficient and costly

Solution Overview

• Standardized SportsBar's raw data into the parent's FMCG star schema
• Built full-load and incremental ETL pipelines using staging tables and Delta MERGE
• Performed a one-time historical backfill of 5 months of missing data
• Automated the full pipeline with Databricks Workflows (task dependencies, retries, failure alerts)
• Delivered a consolidated Gold-layer star schema and an enriched reporting view combining both companies' customer, product, pricing, and sales data

---

Key Business Outcomes

This project transformed disconnected parent and acquired-company data into a trusted, scalable analytics platform. By standardizing data formats, automating incremental pipelines, backfilling five months of missing history, and building a unified Gold layer star schema, the business gained faster reporting, improved decision-making, and a strong foundation for future acquisitions.

---

Tech Stack

Category	Tools / Technologies
Cloud Storage	AWS S3
Data Platform	Databricks
Processing Engine	Apache Spark
Programming	SQL, Python, PySpark, Spark SQL
Storage Format	Delta Lake
Architecture	Bronze, Silver, Gold Medallion Architecture
Data Modeling	Star Schema
Orchestration	Lakeflow Jobs
Incremental Load	Delta Lake MERGE, staging tables
Reporting Layer	Gold analytical view

Project Highlights

• Built an end-to-end ETL pipeline using Databricks and PySpark
• Implemented Medallion Architecture (Bronze → Silver → Gold)
• Automated orchestration using Databricks Lakeflow Jobs
• Supported both historical backfill and incremental processing
• Used Delta Lake MERGE for idempotent upserts
• Consolidated parent and acquired company data into a unified Star Schema
• Generated analytics-ready reporting views for BI tools

Data Architecture Diagram

Design Decisions & Trade-offs

Design Decision	Reason	Trade-off
Medallion Architecture (Bronze → Silver → Gold)	Separates raw, cleaned, and business-ready data, making the pipeline easier to maintain, debug, and extend.	Requires additional storage because data is stored across multiple layers.
Incremental Data Processing	Processes only newly arrived order files, reducing execution time and compute costs compared to full reloads.	Requires staging tables and additional logic to track incremental data.
Delta Lake MERGE for Upserts	Uses ACID-compliant MERGE operations to update existing records and insert new records while preventing duplicates.	MERGE operations are more compute-intensive than simple append operations.
Star Schema for Analytics	Organizes Gold-layer data into fact and dimension tables, improving query performance and simplifying BI reporting.	Introduces some data redundancy compared to a fully normalized model.

---

Data Flow

Data Integration

Data Model

Databricks Lakeflow Jobs

The data integration pipeline is automated using a Databricks Lakeflow Job. The Lakeflow Job orchestrates the execution of multiple notebook tasks in a predefined dependency order, ensuring that all dimension tables are processed before the fact table.

Task Execution Order

Step	Lakeflow Task	Notebook	Purpose
1	processing_dim_customers	1_customer_data_processing.ipynb	Cleans, standardizes, and merges customer data into the parent customer dimension.
2	processing_dim_products	2_products_data_processing.ipynb	Processes product data and merges it into the parent product dimension.
3	processing_dim_price	3_pricing_data_processing.ipynb	Processes gross price data and merges pricing records into the parent pricing dimension.
4	processing_fact_order	2_incremental_load_fact.ipynb	Processes incremental order data and updates the parent fact table.

Lakeflow Job Features

• Automated orchestration using Databricks Lakeflow Jobs
• Sequential task dependencies (depends_on) to ensure correct execution order
• Parameterized notebook execution using catalog and data_source widgets
• Incremental order processing for newly arrived files
• Serverless compute for scalable execution
• Failure email notifications for operational monitoring

---

For Data Catalog CLICK HERE

How to Run the Project CLICK HERE

📁 Project Structure

FMCG Sales Data Integration Pipeline/
│
├── 0_data/
│   ├── 1_parent_company/
│   │   ├── full_load/
│   │   │   ├── dim_customers.csv
│   │   │   ├── dim_products.csv
│   │   │   ├── dim_gross_price.csv
│   │   │   └── fact_orders.csv
│   │   │
│   │   └── incremental_load/
│   │       └── fact_orders.csv
│   │
│   └── 2_child_company/
│       ├── full_load/
│       │   ├── customers.csv
│       │   ├── products.csv
│       │   ├── gross_price.csv
│       │   └── orders.csv
│       │
│       └── incremental_load/
│           └── orders.csv
│
├── 1_code/
│   ├── 1_setup/
│   │   ├── setup_catalog.ipynb
│   │   ├── utilities.ipynb
│   │   └── dim_date_table_creation.ipynb
│   │
│   ├── 2_dimension_processing/
│   │   ├── 1_customer_data_processing.ipynb
│   │   ├── 2_products_data_processing.ipynb
│   │   └── 3_pricing_data_processing.ipynb
│   │
│   ├── 3_fact_processing/
│   │   ├── 1_full_load_fact.ipynb
│   │   └── 2_incremental_load_fact.ipynb
│   │
│   └── 4_sql/
│       └── denormalise_table_query_fmcg.sql
│
├── resources/
│   ├── architecture_diagram.png
│   ├── data_flow_diagram.png
│   ├── data_integration_diagram.png
│   ├── data_model_star_schema.png
│   ├── Data catalog.md
│   └── Project Screen Shoots
│
└── README.md