Enterprise Data Warehouse Sync

Production-Grade ETL Platform for Multi-Source Customer Data Synchronization

Extract · Transform · Validate · Orchestrate · Monitor · Load

Overview

Enterprise Data Warehouse Sync is a production-oriented data engineering platform that synchronizes customer records from multiple SaaS systems into a centralized PostgreSQL warehouse.

The platform automatically extracts customer data from Stripe and Salesforce, validates incoming records against strict schema contracts, transforms heterogeneous schemas into a canonical warehouse model, and loads the final dataset into a centralized, analytics-ready warehouse.

The system is fully containerized, orchestrated through Apache Airflow, monitored through Slack alerts, and continuously validated through GitHub Actions CI/CD pipelines.

Key Features

Multi-source data ingestion from Stripe and Salesforce
Incremental loading using timestamp-based watermark synchronization using source creation timestamps
Pydantic schema validation with strict field-level type enforcement
Automatic retry handling with exponential backoff and jitter
Raw data archival to AWS S3 data lake
Data cleaning and normalization using Pandas
PostgreSQL analytics warehouse — indexed, typed, and query-ready
Idempotent upsert operations — safe to re-run without creating duplicates
Apache Airflow orchestration — scheduled, dependency-aware, and monitored
Slack failure alerting — structured notifications with full execution context
ETL audit logging — per-run record counts, status, and runtime tracking
Warehouse statistics dashboard — operational visibility across all pipeline runs
Dockerized deployment — reproducible across all environments
GitHub Actions CI/CD pipeline — automated quality gates on every push and PR
Automated testing with Pytest — coverage across all pipeline layers

Architecture

The following diagram shows how data flows from source APIs through to the analytics warehouse, with Airflow coordinating every stage.

flowchart LR
    subgraph Sources["Data Sources"]
        A[Stripe API]
        B[Salesforce API]
    end

    subgraph Ingestion["Ingestion & Validation"]
        C[Extraction Layer]
        D[Validation Layer]
        E[Raw Data Storage]
    end

    subgraph Storage["Cloud Storage"]
        F[(AWS S3\nData Lake)]
    end

    subgraph Processing["Transformation"]
        G[Transformation Layer]
        H[Cleaning & Standardization]
        I[Unified Warehouse Schema]
    end

    subgraph Warehouse["Data Warehouse"]
        J[(PostgreSQL)]
    end

    subgraph Ops["Operations"]
        K[Apache Airflow]
        L[Slack Alerts]
        M[GitHub Actions]
    end

    A --> C
    B --> C
    C --> D
    D --> E
    E --> F
    F --> G
    G --> H
    H --> I
    I --> J

    K --> C
    K --> G
    K --> J
    K --> L
    M --> K

ETL Workflow

Each pipeline run executes the following stages in sequence. Airflow enforces task ordering and retries on failure.

flowchart TD
    A([Start])
    A --> B[Extract\nPull raw records from Stripe & Salesforce]
    B --> C[Validate\nEnforce schema contracts via Pydantic]
    C --> D[Transform\nNormalize fields to canonical warehouse format]
    D --> E[Clean\nRemove nulls, duplicates, and malformed entries]
    E --> F[Standardize\nApply uniform types, dates, and naming]
    F --> G[Load\nUpsert records idempotently into PostgreSQL]
    G --> H([Warehouse — Analytics-Ready Data])

    B -- on failure --> Z[Slack Alert]
    C -- on failure --> Z
    D -- on failure --> Z
    E -- on failure --> Z
    F -- on failure --> Z
    G -- on failure --> Z

Stage	Responsibility
Extract	Pull raw records with pagination and retry logic
Validate	Enforce required fields, types, and schema constraints
Transform	Map source fields to the canonical warehouse model
Clean	Handle nulls, deduplication, and malformed values
Standardize	Apply uniform date formats, data types, and naming conventions
Load	Upsert records idempotently using `INSERT ... ON CONFLICT`

Technology Stack

Layer	Technology	Purpose
Language	Python 3.11+	Core pipeline runtime
Data Processing	Pandas, Polars	Transformation and cleaning
Validation	Pydantic	Schema enforcement and type checking
Database	PostgreSQL	Central analytics warehouse
ORM	SQLAlchemy	Database interaction and upsert logic
Orchestration	Apache Airflow	Scheduling, dependency management, retries
Cloud Storage	AWS S3	Intermediate data lake storage
Monitoring	Slack Webhooks	Failure alerts with execution context
Testing	Pytest	Automated test coverage across all layers
DevOps	Docker	Containerized, reproducible infrastructure
CI/CD	GitHub Actions	Automated validation on every push and PR

System Components

Extraction Layer

Connects to external SaaS APIs and retrieves raw customer records reliably.

Responsibilities:

Stripe API integration with cursor-based pagination
Salesforce SOQL query execution via REST API
Exponential backoff retry logic with jitter
Automatic rate limit detection and recovery

Transformation Layer

Reshapes raw source data into the canonical warehouse schema.

Responsibilities:

Schema normalization across heterogeneous sources
Data cleaning and null handling
Metadata enrichment (source, pipeline_version)
Source lineage tagging on every record
Canonical field mapping to the unified customer model

Incremental Synchronization

The pipeline avoids full reloads on every run by tracking the last successful synchronization timestamp per source. Only records created after that timestamp are extracted, transformed, and loaded.

flowchart LR
    A[Stripe API] --> B[Last Sync Check]
    B --> C[New Records Only]
    C --> D[Transform]
    D --> E[Load]

    F[Salesforce API] --> G[Last Sync Check]
    G --> H[New Records Only]
    H --> D

How it works:

Stripe — extraction filters records using created[gt], passing the last recorded sync timestamp as a Unix timestamp parameter.
Salesforce — extraction filters records using a CreatedDate watermark in the SOQL query (WHERE CreatedDate > {last_sync}).
After each successful run, the sync timestamp is updated so the next run picks up only net-new records.

Benefits:

Reduces API call volume significantly on large datasets
Lowers execution time per DAG run
Minimises warehouse write operations and lock contention
Makes the pipeline safe to schedule at high frequency

Warehouse Layer

Persists clean, analytics-ready records into PostgreSQL.

Responsibilities:

Incremental synchronization using timestamp-based watermark synchronization using source creation timestamps
Idempotent upsert operations — safe to re-run without creating duplicates
Indexed, typed, and normalized table structure
Query-ready schema for downstream BI tools

Orchestration Layer

Apache Airflow manages the full DAG lifecycle from scheduling to monitoring.

flowchart TD
    S[Airflow Scheduler\nCron-based trigger]
    S --> T1[extract_task]
    T1 --> T2[transform_task]
    T2 --> T3[load_task]
    T3 --> T4[notify_success]
    T1 -- on_failure_callback --> A[Slack Alert]
    T2 -- on_failure_callback --> A
    T3 -- on_failure_callback --> A

DAG Features:

Cron-based scheduled execution
Task-level dependency enforcement
Configurable retry policies with delay
Live monitoring through the Airflow UI

Observability Layer

Any task failure automatically dispatches a structured Slack alert with full execution context.

sequenceDiagram
    participant Airflow
    participant Callback as on_failure_callback
    participant Slack

    Airflow->>Callback: Task failure detected
    Callback->>Slack: POST /webhooks/...
    Slack-->>Callback: 200 OK
    Note over Slack: Alert includes DAG name, failed task,<br/>timestamp, and run ID

Alert payload includes:

DAG name and failed task identifier
Timestamp of failure
Execution run ID for log lookup
Direct link to Airflow task logs

ETL Audit & Monitoring

Every pipeline run writes a structured record to an internal ETL audit table in PostgreSQL. This provides a persistent, queryable history of all pipeline executions without relying solely on Airflow logs.

Audit table captures:

Column	Description
`source_system`	Origin system (`stripe` or `salesforce`)
`records_extracted`	Total records pulled from the source API
`records_loaded`	Records successfully upserted into the warehouse
`execution_status`	`SUCCESS` or `FAILED`
`runtime_seconds`	Total execution time for the run
`run_timestamp`	UTC timestamp of the pipeline execution

Sample audit log:

Source	Extracted	Loaded	Status	Runtime
Stripe	10	10	SUCCESS	4.2s
Salesforce	5	5	SUCCESS	3.1s

Operational benefits:

Enables trend analysis on pipeline throughput over time
Provides immediate visibility into partial failures (extracted ≠ loaded)
Supports alerting rules based on record count anomalies
Simplifies troubleshooting by correlating Airflow run IDs with warehouse state

CI/CD Pipeline

Every push and pull request automatically triggers the full validation pipeline. No code merges unless all tests pass.

flowchart LR
    A[Developer Push\nor Pull Request]
    A --> B[GitHub Actions\nWorkflow Triggered]
    B --> C[Install Dependencies\npip install -r requirements.txt]
    C --> D[Run Test Suite\npytest tests/ -v]
    D --> E{All Tests Pass?}
    E -->|Yes| F[Merge Allowed]
    E -->|No| G[Pipeline Blocked\nFailures annotated on diff]

Repository Structure

enterprise-data-warehouse-sync/
│
├── .github/
│   └── workflows/
│       └── ci.yml
│           # GitHub Actions CI/CD pipeline
│
├── certs/
│   └── server.crt
│       # Salesforce JWT authentication certificate
│
├── config/
│   # Project configuration files
│
├── dags/
│   └── etl_pipeline_dag.py
│       # Apache Airflow DAG for ETL orchestration
│
├── etl/
│   │
│   ├── extract/
│   │   ├── extract_stripe.py
│   │   │   # Stripe API extraction with pagination support
│   │   │
│   │   └── extract_salesforce.py
│   │       # Salesforce API extraction and SOQL queries
│   │
│   ├── transform/
│   │   ├── transform_pipeline.py
│   │   │   # Main transformation workflow
│   │   │
│   │   ├── transform_stripe.py
│   │   │   # Stripe schema normalization
│   │   │
│   │   ├── transform_salesforce.py
│   │   │   # Salesforce schema normalization
│   │   │
│   │   ├── data_cleaner.py
│   │   │   # Data cleaning and standardization utilities
│   │   │
│   │   └── unified_schema.py
│   │       # Canonical warehouse schema definition
│   │
│   ├── load/
│   │   ├── database.py
│   │   │   # SQLAlchemy engine and session configuration
│   │   │
│   │   ├── warehouse_models.py
│   │   │   # PostgreSQL warehouse table definitions
│   │   │
│   │   └── load_pipeline.py
│   │       # Incremental loading and upsert logic
│   │
│   ├── models/
│   │   ├── stripe_model.py
│   │   │   # Pydantic validation schema for Stripe data
│   │   │
│   │   └── salesforce_model.py
│   │       # Pydantic validation schema for Salesforce data
│   │
│   └── utils/
│       ├── __init__.py
│       │
│       ├── config.py
│       │   # Environment variables and configuration management
│       │
│       ├── logger.py
│       │   # Structured logging utilities
│       │
│       ├── retry_handler.py
│       │   # Retry and rate-limit recovery logic
│       │
│       └── s3_client.py
│           # AWS S3 integration and raw data storage
│
├── tests/
│   ├── testdata/
│   │   └── transformed_stripe_data.json
│   │       # Sample fixture data for automated testing
│   │
│   └── test_transformations.py
│       # Transformation validation test suite
│
├── data/
│   ├── raw/
│   │   # Raw extracted records from source systems
│   │
│   └── processed/
│       # Cleaned and warehouse-ready datasets
│
├── logs/
│   # Pipeline execution logs and debugging output
│
├── .env.example
│   # Example environment variable configuration
│
├── .gitignore
│   # Git ignore rules
│
├── Dockerfile
│   # Application container definition
│
├── docker-compose.yml
│   # Multi-container service orchestration
│
├── requirements.txt
│   # Python dependencies
│
├── LICENSE
│   # MIT License
│
└── README.md
    # Project documentation

Quality Assurance

All records are subject to automated validation at ingestion and pre-load stages.

Check	Scope	Tooling
Required fields present	All source records	Pydantic
Schema consistency	Cross-source field alignment	Custom validators
Data type verification	Field-level type enforcement	Pydantic
Source lineage tracking	Origin system tagged per record	Pipeline metadata
Warehouse readiness	Pre-load schema conformance check	SQLAlchemy
Automated CI/CD execution	Every push and pull request	GitHub Actions + Pytest

Production Engineering Practices

This platform is built around engineering practices that make data pipelines reliable, observable, and maintainable in production environments.

Practice	Implementation
Environment-based configuration	All credentials and parameters managed via `.env` — no hardcoded values anywhere in the codebase
Structured logging	Every pipeline stage emits structured log entries with timestamps, source identifiers, and record counts
Retry and backoff strategy	API calls use exponential backoff with jitter to handle transient failures and rate limits gracefully
Incremental loading	Timestamp-based watermark synchronization using source creation timestamps means only net-new records are processed on each run
Idempotent database writes	`INSERT ... ON CONFLICT DO UPDATE` ensures re-running the pipeline never produces duplicate records
Airflow scheduling and monitoring	DAG-level dependency enforcement, configurable retries, and a live UI for run inspection
Slack alerting	Structured failure notifications include the DAG name, failed task, run ID, and timestamp — enough context to diagnose without opening logs
CI/CD validation	GitHub Actions blocks merges if any Pytest test fails, enforcing quality gates on every contribution
Containerized deployment	Docker Compose defines the full service stack — Airflow, PostgreSQL, and the ETL application — making the environment reproducible across machines
Audit logging	Every pipeline run writes a structured record to the ETL audit table, providing a persistent history of execution status, record counts, and runtime

These practices collectively reduce mean time to recovery (MTTR), eliminate environment-specific bugs, and give operators full visibility into pipeline health without requiring manual log inspection.

Quick Start

Prerequisites

Python 3.11+
Docker and Docker Compose
PostgreSQL (or use the included docker-compose service)
AWS credentials with S3 read/write access
Slack incoming webhook URL

1. Clone the Repository

git clone https://github.com/your-org/enterprise-data-warehouse-sync.git
cd enterprise-data-warehouse-sync

2. Configure Environment

cp .env.example .env

Edit .env with your credentials:

STRIPE_API_KEY=sk_live_...
SALESFORCE_USERNAME=...
SALESFORCE_PASSWORD=...
POSTGRES_URI=postgresql://user:password@localhost:5432/warehouse
AWS_ACCESS_KEY_ID=...
AWS_SECRET_ACCESS_KEY=...
SLACK_WEBHOOK_URL=https://hooks.slack.com/services/...

3. Start the Stack

docker-compose up -d

4. Access Airflow

URL:      http://localhost:8080
Username: airflow
Password: airflow

5. Run the Test Suite

docker-compose exec airflow pytest tests/ -v

Screenshots

Apache Airflow DAG

The ETL workflow is orchestrated using Apache Airflow with task dependencies for extraction, transformation, and loading.

![Airflow DAG](docs/images/airflow_dag.png)

Successful Airflow Execution

A successful DAG run showing all ETL stages completed successfully.

![Airflow Success](docs/images/airflow_success.png)

PostgreSQL Data Warehouse

Warehouse-ready customer records synchronized into PostgreSQL.

![PostgreSQL Warehouse](docs/images/postgres_warehouse.png)

Slack Failure Monitoring

Automatic Slack notifications triggered whenever an Airflow task fails.

![Slack Alert](docs/images/slack_alert.png)

GitHub Actions CI/CD

Automated validation pipeline executed on every push and pull request.

![GitHub Actions](docs/images/github_actions.png)

Docker Infrastructure

Containerized application services running through Docker Compose.

![Docker Containers](docs/images/docker_containers.png)

Key Achievements

Achievement	Detail
Multi-source ETL pipeline	Unified ingestion from Stripe and Salesforce
PostgreSQL data warehouse	Analytics-ready, indexed, and normalized
Apache Airflow orchestration	Scheduled, monitored, and dependency-aware
Slack monitoring and alerting	Zero-surprise failure detection
Dockerized infrastructure	Reproducible across all environments
GitHub Actions CI/CD	Automated quality gates on every merge
Automated testing framework	Pytest coverage across all pipeline layers
Incremental synchronization	Efficient timestamp-based watermark synchronization using source creation timestamps

Contributors

Name	Responsibilities
Harshil Kalsariya	ETL architecture, Airflow DAGs, CI/CD, warehouse design & project coordination
Sravya Maddipati	Data cleaning, transformation logic, schema normalization , CI/CD improvize
Rifaz G	Project improvements, testing
Bhanu Prakash	Documentation & Architecture Learning

License

This project is licensed under the MIT License — see the LICENSE file for full details.

_{Enterprise Data Warehouse Sync · Built for reliable, scalable data engineering}

Name		Name	Last commit message	Last commit date
.github/workflows		.github/workflows
certs		certs
config		config
dags		dags
data		data
docs		docs
etl		etl
tests		tests
.env.example		.env.example
.gitignore		.gitignore
CONTRIBUTING.md		CONTRIBUTING.md
Dockerfile		Dockerfile
Dockerfile.airflow		Dockerfile.airflow
LICENSE		LICENSE
Makefile		Makefile
README.md		README.md
docker-compose.airflow.yml		docker-compose.airflow.yml
docker-compose.yml		docker-compose.yml
requirements-airflow.txt		requirements-airflow.txt
requirements.txt		requirements.txt

Latest commit

History

Folders and files

Repository files navigation

Enterprise Data Warehouse Sync

Table of Contents

Overview

Key Features

Architecture

ETL Workflow

Technology Stack

System Components

Extraction Layer

Transformation Layer

Incremental Synchronization

Warehouse Layer

Orchestration Layer

Observability Layer

ETL Audit & Monitoring

CI/CD Pipeline

Repository Structure

Quality Assurance

Production Engineering Practices

Quick Start

Prerequisites

1. Clone the Repository

2. Configure Environment

3. Start the Stack

4. Access Airflow

5. Run the Test Suite

Screenshots

Apache Airflow DAG

Successful Airflow Execution

PostgreSQL Data Warehouse

Slack Failure Monitoring

GitHub Actions CI/CD

Docker Infrastructure

Key Achievements

Contributors

License

About

Resources

Contributing

Stars

Watchers

Forks

Releases

Packages

Used by

Contributors

Languages