CoreTelecoms Unified Data Platform: The Master Technical Reference

You can download the full PowerPoint presentation using the link below:

Click here to open the PowerPoint presentation

Executive Summary & Project Vision

The CoreTelecoms project delivers a secure, scalable, and fully reproducible ELT pipeline designed to analyze customer complaint and operational experience data.

Primary Goal

Decouple:

Compute: Airflow, Snowflake
Storage: S3

and manage the entire cloud estate through Infrastructure as Code (IaC) using Terraform.

Core Philosophy — “Sanitize-First Approach”

We reject the traditional “dump raw files and clean later” pattern.

Instead, we designed:

A robust Python/Polars extraction layer
Schema normalization before loading
JSON flattening at ingestion time
Strict quality gates before data enters Snowflake

This ensures clean, high-quality Bronze data before it reaches the warehouse.

This is the final installment of the CoreTelecoms Data Platform technical documentation. This section covers Data Lineage, the high-level Data Architecture.

DATA LINEAGE & TRACEABILITY

Objective

Data lineage explains exactly where the data comes from, how it moves, how it changes, and where it ends up. Our platform implements complete, enterprise-grade lineage across Source Systems, Ingestion Scripts, S3, Snowflake, and dbt.

Core Guarantee: Every single record in the final dashboard can be traced back to a specific row in a specific source file from a specific date.

High-Level Lineage Flow

The data flows through a strict pipeline of transformations:

Source Systems: (PostgreSQL CRM, Google Sheets, External S3 Logs)
Ingestion Layer: (Python + Polars + Boto3)
Bronze Layer: (S3 Raw Zone - Parquet)
Silver Layer: (Snowflake RAW -> STAGING -> DEDUP)
Gold Layer: (dbt Dimensions & Facts)
BI Layer: (Power BI / Tableau / SQL Analytics)

Detailed Lineage per Layer

A. Source Systems

PostgreSQL: Tables extracted based on execution_date. Metadata includes the source table name.
S3 Logs: CSV and JSON files. Metadata includes the original key path.
Google Sheets: Sheet ranges. Metadata includes the sheet name.

B. Ingestion Layer (Python)

Every script injects the following metadata columns into the Parquet files:

load_time: The exact timestamp (UTC) when the script processed the row.
source_file: The origin filename (e.g., call_logs_2025_11_25.csv).
row_number: The line number from the source file (critical for debugging bad CSVs).

C. S3 Raw Zone (The Immutable Record)

Structure: s3://coretelecoms-raw/<table>/<YYYY>/<MM>/<DD>/data.parquet
Role: Acts as the persistent history. Even if Snowflake is wiped, the entire warehouse can be rebuilt from these files.

D. Snowflake Loading (Silver)

Flow: S3 -> RAW Table -> STAGING View -> MERGE/SWAP Target.
Added Metadata: Snowflake adds ingested_at (Server timestamp) and file_last_modified.

E. dbt Transformations (Gold)

dbt provides the semantic lineage graph.

Example Chain:

RAW.web_complaints
 -> stg_web_complaints (Cleaned)
    -> fct_web_complaints (Modeled)
       -> fct_unified_complaints (Aggregated)

DATA ARCHITECTURE

Architectural Principles

The CoreTelecoms platform utilizes a Lakehouse + Warehouse Hybrid Architecture.

Decoupling: Compute (Snowflake/Airflow) is strictly separated from Storage (S3).
Idempotency: Re-running any part of the pipeline is safe and deterministic.
Event-Driven: Downstream pipelines wait for data availability (Assets), not arbitrary times.
Governance: Infrastructure is Code (Terraform) and Pipelines are Code (Airflow/dbt).

Architecture Components

A. Ingestion Layer (Bronze)

Tools: Python, Polars, Boto3.
Responsibility: "Sanitize-First." Normalize headers and flatten JSON before storage.

B. S3 Raw Zone (Bronze Persistent)

Structure: Date-partitioned Parquet.
Responsibility: The ultimate source of truth.

C. Snowflake Warehouse (Silver/Gold)

Silver (Staging): Cleaned, typed, and deduplicated data.
Gold (Marts): Star Schema optimized for high-performance BI queries.

D. dbt Layer (Transformation)

Tools: dbt Core, dbt-utils.
Responsibility: Business logic, surrogate key generation, and incremental logic.

E. Airflow Layer (Orchestration)

Tools: Airflow 3.1.3, Astronomer Cosmos.
Responsibility: The "Traffic Controller." Manages retries, backfills (catchup=True), and dependencies.

F. CI/CD Layer

Tools: GitHub Actions.
Responsibility: Validation. Runs Linting, Unit Tests, and Slim CI before deployment.

G. Security Layer

Tools: AWS Secrets Manager, IAM.
Responsibility: Zero-Trust. No hardcoded secrets, encrypted storage, and least-privilege access.

Phase 1: Infrastructure & Environment Provisioning.

I have integrated your draft with the specific technical details from the source text, adding the missing components you requested: Testing Infrastructure, Dependency Management, Virtual Environments, and the Data Lineage foundation.

PHASE 1 — INFRASTRUCTURE & ENVIRONMENT PROVISIONING

1. Objective

The objective of Phase 1 was to establish a secure, scalable, and fully reproducible cloud foundation for the CoreTelecoms Unified Customer Experience Data Platform. This phase ensures the decoupling of compute (Airflow/Snowflake) from storage (S3), managing the entire cloud estate using Infrastructure as Code (IaC).

The entire data platform depends on this foundation. Everything from Airflow orchestration to Snowflake warehousing and S3 storage is created, governed, and version-controlled through Terraform.

2. Architectural Decisions & Thought Process

A. Infrastructure as Code (Terraform)

Decision: We chose Terraform over manual AWS console steps or CloudFormation.
Rationale:
- Reproducibility: The entire environment (AWS & Snowflake) can be destroyed and recreated with a single command (terraform apply).
- Version Control: Infrastructure configuration is stored in Git, enabling peer review and history tracking.
- Provider Support: Terraform supports both AWS and Snowflake providers natively, allowing unified management of the stack.

B. State Management Strategy

Decision: Implemented Remote State (S3) with State Locking (DynamoDB).
Rationale:
- Corruption Prevention: Storing state locally (terraform.tfstate) is a risk for corruption and prevents collaboration.
- Concurrency Control: DynamoDB locking prevents race conditions if CI/CD pipelines and a developer attempt to deploy simultaneously.

C. Data Warehouse Selection – Snowflake

Decision: Selected Snowflake over AWS Redshift.
Rationale:
- Decoupled Architecture: Snowflake's separation of Storage and Compute allows us to auto-suspend the warehouse after 60 seconds (auto_suspend = 60). This keeps costs near zero during idle times while maintaining high performance for loading.
- Zero-Copy Cloning: Critical for the "Swap Pattern" used in production deployments to ensure zero downtime.
- Semi-Structured Data: Native support for JSON/Parquet variants was essential for our "Sanitize-First" ingestion strategy.

D. Containerization Strategy

Decision: Extended the official apache/airflow:3.1.3 image to create a custom Docker image.
Rationale:
- Dependency Consistency: We avoid runtime installation of heavy libraries. By baking dependencies into the image, we ensure that if code runs on a developer's machine, it runs in production.
- Security: We specifically removed non-essential tools to maintain a smaller, more secure footprint.

E. Secrets Management Hardening

Decision: Migrated from local .env files to AWS Secrets Manager.
Rationale:
- Encryption: .env files store plaintext data; Secrets Manager encrypts data at rest using AWS KMS.
- Access Control: Access is controlled by granular IAM Policies rather than weak file permissions.
- Compliance: Operations are fully logged in AWS CloudTrail for auditing.

3. Implementation Log: Cloud & Security

A. Security Architecture

To connect Snowflake to AWS S3 securely without using vulnerable long-lived access keys, we implemented a Storage Integration.

The Mechanism: Snowflake assumes a specific AWS IAM Role to access the S3 Data Lake.
The Handshake: A circular dependency exists where Terraform creates the Role, but AWS requires the Snowflake-generated User ARN to trust it. We resolved this by accepting a manual step to update the AWS Trust Policy with the Snowflake output.

B. Secrets Inventory

We provisioned the following high-value secrets in AWS Secrets Manager, using CoreTelecoms/ as a secure namespace:

${var.project_name}/postgres_credentials: Host, user, and password for the source database.
${var.project_name}/snowflake_credentials: Account, user, warehouse, and password.
${var.project_name}/google_service_account: The full JSON key for Google Sheets authentication.
${var.project_name}/source_aws_credentials: Keys for reading the external source S3 bucket.
${var.project_name}/target_aws_credentials: Keys for writing to our internal Data Lake.

C. Resource Tagging & Governance

Tagging Strategy: The AWS Provider is configured to automatically tag all resources with Project = CoreTelecoms.
Benefit: This ensures precise cost allocation and ownership visibility across the cloud estate.

D. Snowflake Infrastructure

Terraform successfully provisioned:

Database: CORETELECOMS_DW.
Schemas: RAW (Ingestion) and ANALYTICS (Transformation).
Warehouse: CORETELECOMS_WH (Renamed from default to ensure isolation).

4. Local Development Environment & Dependencies

We architected a Hybrid Development Environment to balance speed with consistency.

A. Python Virtual Environment (`venv`)

Usage: Used for lightweight local tooling that doesn't require the heavy Airflow runtime.
Tools Installed: Terraform CLI, dbt CLI (for local debugging), and linters (SQLFluff).
Benefit: Allows for rapid terraform apply or dbt run commands without the overhead of shelling into a Docker container.

B. Docker & Airflow Environment

Configuration: The docker-compose.yaml mounts local directories (/scripts, /dags, /dbt_project) directly into the container.
Hot Reloading: This enables rapid iteration—editing a Python script or SQL model locally is instantly reflected in the running Airflow instance without a rebuild.

C. Requirements & Dependency Management

We managed dependencies via a custom Dockerfile to solve "Dependency Hell".

System Dependencies: Installed build-essential and git to support dbt compilation.
Python Libraries (The Image):
- polars: For high-performance, memory-efficient ETL (replacing Pandas).
- dbt-snowflake: To enable native transformation logic within the container.
- astronomer-cosmos: To render dbt projects as Airflow TaskGroups.
- apache-airflow-providers-*: For AWS, Snowflake, and Google integrations. e.t.c.

D. Data Lineage Foundation

Setup: By installing astronomer-cosmos and dbt-core in the infrastructure phase, we laid the groundwork for end-to-end lineage.
Visibility: This setup ensures that every transformation task is traceable back to its source DAG, providing a clear dependency graph from "Ingestion" to "Dashboard" directly in the Airflow UI.

5. Testing Infrastructure

To ensure a fail-fast development cycle, we provisioned a dual-layer testing infrastructure.

A. Test Configuration

Tooling: pytest is installed in the development environment.
Mocking: We use moto to mock AWS S3 interactions and unittest.mock for Google Sheets.
Rationale: Relying on real cloud services for tests is slow and costly. Mocking allows us to simulate errors (e.g., missing files) instantly and deterministically.

B. Test Scripts Structure

We organized tests into two distinct categories:

Logic Tests (tests/test_extraction_logic.py): Validates individual Python functions (e.g., parsing JSON) in isolation.
Integrity Tests (tests/test_dag_integrity.py): Validates Airflow objects, ensuring DAGs import without errors and have valid dependencies.

The platform infrastructure is now fully provisioned, secured, and ready for the orchestration and ingestion pipelines.

This is the comprehensive Phase 2: Airflow Orchestration Environment documentation, detailing the heart of your data platform. I have integrated your architectural decisions, specific implementation details, and project structure into a cohesive technical reference.

PHASE 2 — AIRFLOW ORCHESTRATION ENVIRONMENT

2.1. Overview & Objective

Phase 2 transforms Apache Airflow from a simple task scheduler into the central control plane for the entire CoreTelecoms Data Platform. It coordinates the complete data lifecycle—from ingestion and loading to transformation and testing—while ensuring full observability and resilience.

Primary Objectives:

Production-Grade Reliability: Pipelines must run predictably, handling failures gracefully without manual intervention.
Full Lineage Visibility: Tracking data flow from Source → S3 (Bronze) → Snowflake Raw (Silver) → dbt Marts (Gold).
Event-Driven Execution: Utilizing Airflow Assets (Datasets) to decouple DAGs and trigger downstream processes only when data is ready.
Reproducibility: Ensuring every run is deterministic, with robust support for backfilling historical data.

2.2. Architectural Decisions

A. Custom "Image" Strategy

Instead of relying on the lightweight official image, we built a custom Docker image extending apache/airflow:3.1.3.

Rationale: "Dependency Hell" is a major risk in production. By baking heavy libraries directly into the image, we ensure consistency across all environments (Local, CI, Production).
Key Components Installed:
- Data Processing: polars (for high-performance ETL), pyarrow.
- Transformation: dbt-snowflake, dbt-core.
- Orchestration: astronomer-cosmos (for native dbt integration).
- Cloud Providers: apache-airflow-providers-amazon, google, snowflake.
- Utilities: boto3 (AWS SDK), connectorx (Fast DB extraction).

B. Declarative Configuration (Infrastructure as Code)

We rejected manually configuring Connections in the Airflow UI to prevent configuration drift.

Implementation: Connections are injected via environment variables in the .env file.
- AIRFLOW_CONN_SNOWFLAKE_DEFAULT=snowflake://...
- AIRFLOW_CONN_AWS_DEFAULT=aws://...
Benefit: A new engineer can clone the repo, run docker-compose up -d, and the entire environment auto-configures instantly.

C. Hybrid Credential Strategy

We employ a "Dual Credential" model to separate orchestration concerns from execution logic.

Airflow Operators: Use standard Connection URIs (e.g., snowflake_conn_id) for tasks like SQLExecuteQueryOperator.
Python Scripts: Use raw environment variables (AWS_ACCESS_KEY_ID, GOOGLE_APPLICATION_CREDENTIALS) for libraries like boto3 and polars.

Why: Libraries like boto3 and gspread do not natively understand Airflow Connection objects. This separation keeps scripts simple and testable outside of Airflow.

D. "Hot Reload" Development Environment

The docker-compose.yaml is configured with volume mounts for rapid iteration:

./dags:/opt/airflow/dags
./scripts:/opt/airflow/scripts
./dbt_core_telecoms:/opt/airflow/dbt_core_telecoms
Benefit: Modifying a DAG, Python script, or SQL model locally updates the running container immediately. No rebuilds are required for code changes.

2.3. Idempotency Architecture

Idempotency is the cornerstone of our pipeline reliability: "Re-running a successful or failed task must never corrupt data."

A. Ingestion Idempotency

Deterministic S3 Paths: Scripts write to s3://bucket/table/YYYY/MM/DD/data.parquet. Re-running a day simply overwrites the exact same object.
Local State Reset: Scripts clean the /tmp directory before execution to prevent residual data contamination.
Strict Exit Codes: Scripts raise sys.exit(1) on failure, ensuring Airflow marks the task as failed and halts downstream dependencies.

B. Snowflake Loading Idempotency

Static Tables (Swap Pattern):
- Logic: CLONE Prod to Transient -> TRUNCATE -> LOAD -> SWAP.
- Result: Atomic replacement. The table never appears empty, and failures leave production untouched.
Daily Tables (Merge Pattern):
- Logic: Load to Staging -> DEDUP (using QUALIFY ROW_NUMBER()) -> MERGE into Final.
- Result: You can run the load job 10 times for the same day; Snowflake guarantees exactly one unique record per ID.

C. Transformation Idempotency

Incremental Models: dbt uses WHERE ingested_at > (SELECT MAX(ingested_at) ...) to process only new data.
Result: Re-runs are safe and efficient, avoiding duplicate processing.

2.4. Orchestration Features & Reliability

A. Event-Driven Execution (Assets)

We moved away from strict time-based scheduling to a reactive Dataset-Driven model.

Producer: ingestion_pipeline runs daily and updates the S3_RAW_DATA_READY asset.
Consumer: load_snowflake_pipeline triggers automatically when this asset updates.
Downstream: transform_dbt_pipeline triggers on SNOWFLAKE_RAW_READY.
Benefit: Eliminates "blind" scheduling. Downstream tasks never run until upstream data is confirmed ready.

B. Backfilling & "Time Travel"

Configuration: catchup=True and start_date=2025-11-20.
Capability: Airflow automatically schedules runs for all past dates from the start date to the present.
Use Case: Allows us to re-process historical data (e.g., after fixing a bug) simply by clearing the DAG's status.

C. `LatestOnlyOperator`

Problem: During a backfill of 10 days, we don't want to download the same static "Agents" file 10 times.
Solution: We wrap static tasks with the LatestOnlyOperator.
Result: These tasks run only for the most recent execution date, skipping redundant work during backfills.

D. Retry Strategy & SLAs

Airflow Level: retries=3, retry_delay=timedelta(minutes=5). Handles transient cloud hiccups (S3 timeouts, API limits).
Script Level: Polars scripts implement internal retry loops for HTTP downloads to handle network blips gracefully.

E. Failure Notifications

System: We implemented notifications.py to handle alerts.
Triggers:
- On Failure: Sends critical alerts (Email/Slack) to engineering.
- On SLA Breach: Alerts if a pipeline runs longer than expected.

2.5. Project Setup Guide

This guide allows any engineer to recreate the CoreTelecoms environment from scratch.

Step 1: Clone & Configure

git clone https://github.com/CoreTelecoms/coretelecoms-data-platform.git
cd coretelecoms-data-platform
# Create .env file with AWS, Snowflake, and Postgres credentials

Step 2: Infrastructure Provisioning

cd terraform/
terraform init
terraform plan
terraform apply
# Outputs S3 Bucket Name and IAM Role ARNs

Step 3: Launch Orchestration

docker-compose up --build -d
# Airflow UI available at http://localhost:8080

Step 4: Initialize Backfill

Enable the ingestion_pipeline DAG in the UI.
Due to catchup=True, it will immediately begin processing data from 2025-11-20.

2.6. Full Project Structure

The project is organized to enforce strict separation of concerns.

CORETELECOMS-DATA-PLATFORM/
├── .github/
│   └── workflows/              # CI/CD Pipelines (Linting, Testing, Deployment)
│       ├── ci_pipeline.yml
│       └── cd_pipeline.yml
├── config/                     # Static configuration files
│   ├── airflow.cfg
│   └── google_credentials.json
├── dags/                       # Airflow Orchestration Logic
│   ├── assets.py               # Dataset definitions for event-driven triggers
│   ├── ingestion_pipeline.py   # Producer DAG (Source -> S3)
│   ├── load_snowflake_pipeline.py # Consumer DAG (S3 -> Snowflake)
│   ├── notifications.py        # Alerting utilities
│   └── transform_dbt_pipeline.py # Transformation DAG (Cosmos/dbt)
├── dbt_core_telecoms/          # dbt Transformation Project
│   ├── dbt_project.yml
│   ├── models/                 # Staging (Silver) and Marts (Gold) SQL models
│   └── packages.yml
├── scripts/                    # Pure Python Extraction Logic (Process Isolated)
│   ├── extract_gsheets.py
│   ├── extract_postgres.py
│   └── extract_s3_data.py
├── snowflake_sql/              # Setup scripts for Snowflake RBAC/Warehouses
├── terraform/                  # Infrastructure as Code
│   ├── backend.tf              # S3 Remote State config
│   ├── provider.tf             # AWS & Snowflake provider setup
│   ├── resources.tf            # Main resource definitions
│   └── secrets.tf              # Secrets Manager configuration
├── tests/                      # Dual-Layer Testing Suite
│   ├── test_dag_integrity.py   # Validates Airflow structure
│   ├── test_extraction_logic.py # Validates Python logic (Unit Tests)
│   └── test_load_snowflake_pipeline.py
├── .env                        # Local Environment Variables (Gitignored)
├── docker-compose.yaml         # Local Airflow Runtime configuration
└── Dockerfile                  # Custom "Fat" Airflow Image definition

PHASE 3 — DATA INGESTION & ADVANCED ORCHESTRATION

3.1. Objective of the Ingestion Layer

The ingestion layer is the gateway to the data platform. Its primary mandate is to decouple source systems from the warehouse, ensuring that only valid, schema-compliant data reaches Snowflake.

Core Objectives:

High-Speed Extraction: Utilizing connectorx and polars to extract data from Postgres, S3, and Google Sheets efficiently.
Schema Standardization: converting "messy" source headers (e.g., iD, DATE of biRTH) into standardized snake_case before storage.
Auditability: Every single row must be stamped with a load_time to prove when it entered the data lake.
Parquet Storage: Storing data in the S3 Raw Zone using open-format Parquet files for compression and type preservation.
Idempotency: Ensuring that re-running an ingestion job for a past date safely overwrites the specific partition without creating duplicates.
Event-Driven Triggering: Automatically waking up downstream Snowflake loaders via Airflow Assets.

3.2. Ingestion Architecture — High-Level Flow

The architecture follows a strictly decoupled Producer-Consumer pattern.

Step-by-Step Execution:

Trigger: Airflow triggers the ingestion_pipeline DAG (via Schedule, Manual Run, or Backfill).
Parallel Extraction: The daily_ingestion TaskGroup spins up parallel tasks for Postgres, S3, and Google Sheets.
Process Isolation: The BashOperator triggers isolated Python scripts in the scripts/ directory.
- Action: Scripts download raw data to a local temp directory (/tmp).
- Action: Polars processes the data (Normalization, Flattening JSON, Conversion to Parquet).
- Action: Boto3 uploads the sanitized Parquet file to the S3 Target Bucket.
Asset Emission: Upon successful completion of all tasks, Airflow updates the S3_RAW_DATA_READY dataset asset.
Downstream Trigger: The load_snowflake_pipeline DAG detects the asset update and begins loading.
Transformation: Once loading is complete, the transform_dbt_pipeline executes dbt models.

3.3. Detailed Breakdown of Ingestion Sources

The platform ingests from three distinct source systems, each requiring specific handling strategies.

A. PostgreSQL CRM Database

Data Content: Dynamic web form requests and ticket interactions.
Target Table Logic: The source system creates new tables daily (e.g., web_form_request_2025_11_25). The script dynamically selects the correct table based on the Airflow execution_date.
Tooling: connectorx is used for high-speed, parallel extraction, significantly outperforming standard pandas read_sql.
Lineage Captured: Source Table Name, Extract Timestamp, Execution Date.

B. AWS S3 External Buckets (Customer Experience Logs)

Data Content: High-volume CSV Call Logs and nested JSON Social Media complaints.
The "Column-Oriented" JSON Challenge:
- Problem: The source JSON data was structured as a dict-of-dicts (Column-Oriented) rather than a list of records (Row-Oriented).
- Solution: We implemented custom Polars logic in extract_s3_data.py to transpose and flatten these nested structures into a tabular format before writing to Parquet.
Tooling: boto3 (Download/Upload) and polars (JSON Flattening/Parquet Conversion).

C. Google Sheets (Reference Dimensions)

Data Content: Agent metadata, Customer master data, and Categorization rules.
Tooling: Google Service Account (JSON Key) and gspread.
Process:
- Authenticate via Service Account.
- Read specific sheet ranges.
- Header Normalization: Crucial step to convert iD or NamE to id and name to prevent Snowflake case-sensitivity errors.
- Write to Parquet -> Upload to S3.

3.4. Ingestion Scripts (Full Breakdown)

All ingestion logic is encapsulated in modular scripts within scripts/, ensuring separation from orchestration logic.

A. `extract_postgres.py`

Function: Connects to the remote Postgres instance.
Key Logic: Calculates the target table name (Web_form_request_YYYY_MM_DD) using the passed execution date argument.
Safety: Implements sys.exit(1) if the specific daily table is missing, ensuring the pipeline fails loudly rather than silently succeeding.

B. `extract_s3_data.py`

Function: A unified script handling both CSV and JSON formats.
Key Logic:
- Dual Credential Context: Authenticates to the Source bucket using one set of keys and the Target Data Lake bucket using another.
- Path Mapping: Handles inconsistencies in source folder naming (e.g., mapping call logs/ to call_logs).
- Artifact Removal: Strips Pandas index artifacts (like __index_level_0__) before saving.

C. `extract_gsheets.py`

Function: Extracts reference data from Google Sheets.
Key Logic:
- Mounts the Google Credentials JSON file into the container at runtime.
- Validates that required columns exist before processing.
- Adds a load_time timestamp to every row.

3.5. S3 Raw Zone Structure

Data is stored in the S3 Raw Zone using a strict Date-Partitioned Directory Layout.

Structure: s3://coretelecoms-raw/<table>/<YYYY>/<MM>/<DD>/data.parquet

Advantages:

Idempotency: Re-running the job for 2025-11-20 overwrites only that specific folder. It is physically impossible to create duplicate files for the same day in different locations.
Incremental Loading: Snowflake can easily target specific partitions using the directory pattern.
Safe Backfills: We can re-process historical data without touching "Today's" data.

3.6. Airflow Orchestration of Ingestion

The orchestration strategy focuses on parallelism and isolation.

TaskGroups: We group independent tasks (extract_postgres, extract_s3_call_logs, extract_s3_social) into a daily_ingestion TaskGroup. This ensures that a delay in Postgres extraction does not block the downloading of S3 logs.
BashOperator: We use BashOperator to execute the Python scripts.
- Why? This provides Process Isolation. If Polars consumes 8GB of RAM processing a massive file, the memory is reclaimed immediately after the script finishes. It prevents the Airflow Worker process from bloating and crashing.
Strict Dependencies: The emit_asset task is strictly downstream of the TaskGroup. The asset S3_RAW_DATA_READY is only emitted if all ingestion tasks succeed.

3.7. Error Handling & Re-Runs

Reliability is enforced through a "Fail Fast" philosophy.

Script Level:
- If a file is missing -> Error (Exit Code 1).
- If JSON parsing fails -> Error (Exit Code 1).
- If schema is empty -> Error (Exit Code 1).
Orchestration Level:
- Retries: configured to retries=3 with a 5-minute delay to handle transient network issues.
- Alerting: On final failure, notifications.py sends an alert.
Recovery: Because the S3 writes are idempotent, recovering from a failure is as simple as clicking "Clear Task" in Airflow. The script will simply re-download and re-overwrite the partition.

3.8. Downstream Triggering — Dataset Asset (CRUCIAL)

We moved away from time-based scheduling for inter-DAG dependencies.

The Signal: S3_RAW_DATA_READY
Mechanism: When the ingestion_pipeline completes, it updates this asset.
Response: The load_snowflake_pipeline is scheduled on [S3_RAW_DATA_READY].
Business Value:
- Cost Savings: Snowflake never wakes up if Ingestion fails.
- Timeliness: Loading starts immediately after data arrives, rather than waiting for an arbitrary time (e.g., 2:00 AM).

3.9. Backfills & Historical Ingestion

The architecture natively supports "Time Travel".

Configuration: catchup=True, start_date=datetime(2025, 11, 20).
Execution: When deployed, Airflow automatically scheduled runs for Nov 20, 21, 22, and 23.
Static Data Handling: The LatestOnlyOperator was implemented for static datasets (Customers/Agents) to ensure they were not redundantly downloaded 4 times during the backfill, while daily logs were processed for every single historical day.

3.10. End-to-End Ingestion Lineage

           +------------------------+
           |   PostgreSQL CRM      |
           +-----------+------------+
                       |
                       v
 +---------+    +-------------+    +---------------------------+
 | Google  |    |  AWS S3     |    |    External Data Logs     |
 | Sheets  |    |  (CSV/JSON) |    |      (JSON / CSV)          |
 +----+----+    +------+------|    +---------------+-------------+
      \              |                           /
       \             |                          /
        \            v                         /
         +------------------------------------+
         |      Python Polars Ingestion        |  <-- "Sanitize-First" Strategy
         |    (Normalization & Flattening)    |
         +------------------------------------+
                        |
                        v
        +-------------------------------------+
        |  S3 Raw Zone (date-partitioned)     |  <-- Immutable System of Record
        | s3://.../table/YYYY/MM/DD/data.parquet |
        +-------------------------------------+
                        |
                        v
        Airflow emits S3_RAW_DATA_READY Asset  <-- Triggers Snowflake Load

This is the fully detailed Phase 4: Snowflake Loading Pipeline documentation. I have expanded it to include your specific SQL implementation patterns, warehouse configuration, and idempotency logic.

PHASE 4 — SNOWFLAKE LOADING PIPELINE

4.1. Overview & Objective

This phase manages the critical transition from the Bronze Layer (Raw S3 Parquet) to the Silver Layer (Structured Snowflake Tables). It is not a simple copy command; it is an intelligent loading system designed to handle data quality issues, duplicates, and schema drift before the data is ready for transformation.

Core Objectives:

Atomic Operations: Using SWAP and transactions to prevent partial loads or broken tables during updates.
Idempotency: Implementing logic that allows the same file to be loaded ten times without creating a single duplicate record.
Incrementalism: Using MERGE strategies to upsert data efficiently, processing only what has changed.
Traceability: Adding lineage columns (load_time, file_name) to every row for audit purposes.
Event-Driven Triggering: This pipeline is triggered automatically by the S3_RAW_DATA_READY dataset asset, ensuring no idle warehouse time.

4.2. Snowflake Architecture Overview

The Snowflake environment is structured using the Medallion Architecture, optimized for the "Sanitize-First" strategy.

1. RAW Layer (The Landing Zone)

Purpose: Mirrors the S3 Parquet structure exactly.
State: Transient / Temporary.
Function: This is where raw data lands first. We use it to apply deduplication logic (QUALIFY ROW_NUMBER) before moving data to permanent storage.

2. STAGING Layer (The Clean Zone)

Purpose: Holds clean, typed, and deduplicated data.
State: Persistent (but rebuildable).
Function: Standardizes column names (e.g., date_of_birth vs DOB), enforces data types (String -> Timestamp), and serves as the source for dbt models.

3. ANALYTICS Layer (The Gold Zone)

Purpose: Business-ready dimensional models.
State: Persistent & Optimized.
Function: Contains Star Schema tables (FACT_CALLS, DIM_CUSTOMERS) ready for BI tools. (Managed by Phase 5: dbt).

4.3. Two Major Load Patterns

The pipeline implements two distinct loading strategies based on the nature of the data (Static vs. Transactional).

A. Pattern 1: Static Tables (The Swap Pattern)

Target Datasets: DIM_AGENTS, DIM_CUSTOMERS, DIM_COMPLAINT_CATEGORIES.

The Challenge: Users expect these reference tables to be 100% complete at all times. We cannot TRUNCATE the live table, or reports will fail while the load is running.

The Solution: Blue/Green Deployment (Swap)

Clone: Create a temporary clone of the production table (_CLONE).
Truncate: Empty the clone (Metadata operation, near-instant).
Load: COPY INTO the clone from S3.
Atomic Swap: Execute ALTER TABLE ... SWAP WITH ....
- Result: The switch is instantaneous. Zero downtime. If the load fails at step 3, the swap never happens, and Production remains untouched.

B. Pattern 2: Daily Tables (The Merge Pattern)

Target Datasets: FCT_CALL_LOGS, FCT_WEB_COMPLAINTS, FCT_SOCIAL_MEDIA.

The Challenge: These tables receive new rows every day. We cannot reload the whole history (too expensive). We must handle updates (status changes) and new inserts (new tickets).

The Solution: Incremental Merge

Staging: Load raw data into a temporary Staging table.
Deduplicate: Filter duplicates from the batch.
Merge: Run a standard SQL MERGE command to Upsert (Update existing + Insert new) into the Final table.

4.4. Deduplication Logic (The "Row Number" Pattern)

Raw data is often messy. We might receive the same "Call Log" in two different files due to an upstream retry. To handle this, we apply strict deduplication before the final load.

The Logic: We use Snowflake's window function QUALIFY to select only the most recent version of a record.

SELECT
    complaint_id,
    customer_id,
    status,
    load_time
FROM RAW.COMPLAINTS_STAGING
QUALIFY ROW_NUMBER() OVER (
    PARTITION BY complaint_id  -- The Natural Key
    ORDER BY load_time DESC    -- Keep the latest version
) = 1;

Outcome: Even if we ingest the same file 5 times, this logic ensures only 1 unique record per complaint_id enters the Analytics layer.

4.5. The MERGE Logic (Upsert Strategy)

For transactional tables, we use the MERGE statement to handle data evolution.

MERGE INTO ANALYTICS.FCT_COMPLAINTS F
USING (
    SELECT * FROM RAW.COMPLAINTS_STAGING_DEDUPED
) S
ON F.complaint_id = S.complaint_id

-- If the ID exists, update the status (e.g., Open -> Closed)
WHEN MATCHED THEN UPDATE SET
    F.status = S.status,
    F.updated_at = CURRENT_TIMESTAMP()

-- If the ID is new, insert the row
WHEN NOT MATCHED THEN INSERT (
    complaint_id, status, ...
) VALUES (
    S.complaint_id, S.status, ...
);

Lineage Benefit: This approach preserves the history of the record while keeping the current state accurate.

4.6. Warehouse Strategy & Cost Control

We optimized the compute resources to ensure the project stays within budget while handling heavy loads.

Warehouse: CORETELECOMS_WH
Size: X-SMALL (Lowest cost tier).
Auto-Suspend: 60 Seconds.
- Effect: The warehouse shuts down immediately after the load finishes. We pay exactly for the seconds used, often costing pennies per run.
Auto-Resume: TRUE.
- Effect: The warehouse wakes up automatically when Airflow sends a query.

4.7. Observability & Query Tagging

To assist with debugging and auditing, every query run by the pipeline is tagged.

Implementation:

ALTER SESSION SET QUERY_TAG = 'CoreTelecoms_Load_2025-11-26';

Usage:

We can query Snowflake's QUERY_HISTORY view filtering by this tag.
Allows us to instantly find failed queries or performance bottlenecks associated with a specific daily run.

4.8. Data Lineage & Traceability

We add specific metadata columns during the load to ensure complete traceability from S3 to Snowflake.

Column Name	Source	Purpose
`load_time`	Python (Polars)	When the data was extracted from the source.
`ingested_at`	Snowflake (Default)	When the data landed in the warehouse.
`file_name`	Metadata	Which specific S3 file contributed this row.
`file_row_number`	Metadata	The exact line number in the source file.

Result: If a bad record appears in the dashboard, we can trace it back to the exact S3 file and line number that created it.

4.9. Orchestration (Airflow DAG)

The loading process is orchestrated by load_snowflake_pipeline.py.

Trigger: S3_RAW_DATA_READY (Dataset Asset).
Operator: SQLExecuteQueryOperator (Generic SQL execution).
Flow:
1. Sensor: Verify S3 data availability.
2. Parallel Loading: Trigger load_static_tables (Swap) and load_daily_tables (Merge) task groups simultaneously.
3. Validation: Check row counts.
4. Asset Emission: Emit SNOWFLAKE_RAW_READY to wake up the transformation pipeline.

This is the fully detailed Phase 5: Transformation Layer documentation. I have expanded your draft to include specific implementation details regarding the Medallion Architecture, the UNION ALL strategy for the unified fact table, and the technical configuration of Astronomer Cosmos.

This phase represents the "T" in ELT, converting raw Snowflake tables into high-value business assets.

PHASE 5 — TRANSFORMATION LAYER (DBT + COSMOS)

5.1. Objective of the Transformation Layer

The Transformation Layer acts as the bridge between raw data and business intelligence. While the Ingestion layer ensures data arrives safely, this layer ensures it makes sense.

Core Objectives:

Standardization: Converting disparate source schemas (Postgres, S3 CSV, JSON) into a unified naming convention.
Business Logic Application: enforcing rules (e.g., "If Agent ID is null, map to 'Unassigned'").
Star Schema Construction: Building highly optimized Fact and Dimension tables for BI tools like Tableau/PowerBI.
Incremental Processing: processing only new rows to keep Snowflake compute costs low.
Orchestration Visibility: Using Astronomer Cosmos to render individual dbt models as visible, retriable tasks within Airflow, rather than a black-box execution.

5.2. Medallion Architecture (Silver → Gold)

The dbt project implements a strict Medallion Architecture to organize data quality levels.

1. Bronze Layer (Raw)

Location: Snowflake RAW Schema.
State: Raw Parquet data loaded by Phase 4.
Characteristics: Immutable, historical, and exactly matches the source system structure.

2. Silver Layer (Staging)

Location: dbt_core_telecoms/models/staging/ -> Snowflake STAGING Schema.
Type: Transient Views/Tables.
Responsibilities:
- Renaming: Mapping specific source IDs (e.g., call_id, request_id) to a unified source_complaint_id.
- Casting: Converting string timestamps to TIMESTAMP_NTZ.
- Sanitization: Handling nulls (e.g., defaulting duration_seconds to 0 for social media).

3. Gold Layer (Marts)

Location: dbt_core_telecoms/models/marts/ -> Snowflake ANALYTICS Schema.
Type: Persistent Tables.
Responsibilities:
- Dimensions: Context (Who, What, Where).
- Facts: Measurements (How long, How many).
- Aggregations: The "Unified Complaint" view combining all channels.

5.3. dbt Project Structure (Full)

The project follows a modular structure to separate concerns between "cleaning" (Staging) and "modeling" (Marts).

dbt_core_telecoms/
├── dbt_project.yml             # Project config & model materialization rules
├── packages.yml                # Dependencies (dbt-utils)
├── profiles.yml                # (Injected dynamically by Cosmos)
└── models/
    ├── staging/                # The Silver Layer
    │   ├── stg_call_logs.sql
    │   ├── stg_web_complaints.sql
    │   ├── stg_social_media.sql
    │   ├── stg_customers.sql
    │   ├── stg_agents.sql
    │   └── sources.yml         # Source definitions & Data Contracts
    │
    └── marts/                  # The Gold Layer
        ├── dimensions/
        │   ├── dim_customers.sql
        │   ├── dim_agents.sql
        │   └── dim_channels.sql
        │
        └── facts/
            ├── fct_call_logs.sql
            ├── fct_web_complaints.sql
            └── fct_unified_complaints.sql

5.4. Staging Layer (Silver Layer)

The staging layer protects the downstream models from raw data inconsistencies. We configured dbt_project.yml to materialize these as Transient Tables to save storage costs (no Time Travel required for staging).

Key Transformations Implemented:

ID Unification:
- Logic: RENAME call_id AS source_complaint_id.
- Why: Allows unioning different sources later.
Agent Normalization:
- Logic: COALESCE(agent_id, '-1').
- Why: Social media complaints may have no agent initially. We map them to a dummy "Unassigned" agent in the Dimension table to preserve referential integrity.
Timestamp Standardization:
- Logic: TRY_TO_TIMESTAMP(created_at).

5.5. Marts Layer (Gold Layer)

This layer implements the Star Schema optimized for OLAP performance.

A. Dimension Tables

dim_customers: Unique customer profile, cleaned names, and location data.
dim_agents: Agent metadata (Name, Tier). Includes row -1 for "Unassigned".
dim_channels: Lookup table for channel types (Web, Call, Social).

B. Fact Tables

fct_unified_complaints (The Core Model):
- Strategy: Uses a UNION ALL approach to stack Call Logs, Web Forms, and Social Media into a single tall table.
- Benefit: Enables "Omnichannel Analysis"—managers can see support volume across all channels in one dashboard without joining 3 different tables.

5.6. Surrogate Key Strategy

We utilize the dbt-utils package to generate robust keys.

Problem: Natural keys collide. A call_id = 100 and a web_request_id = 100 are different events, but look identical to the database.
Solution: We generate a Surrogate Key (complaint_pk).

Implementation:

{{ dbt_utils.generate_surrogate_key(['source_complaint_id', 'channel_type']) }}

Result: A statistically unique hash that guarantees integrity across the unified fact table.

5.7. Incremental Processing (Cost Optimization)

To handle scale, the main fact table fct_unified_complaints is configured as an Incremental Model.

Configuration:

{{
    config(
        materialized='incremental',
        unique_key='complaint_pk'
    )
}}

SELECT ...
FROM ...
{% if is_incremental() %}
  -- Only process rows that arrived since the last run
  WHERE ingested_at > (SELECT MAX(ingested_at) FROM {{ this }})
{% endif %}

Benefits:

Speed: Processing 5,000 daily rows takes seconds, vs. processing 5 million historical rows.
Cost: Drastically reduces Snowflake compute credits.

5.8. Data Tests (Quality Gates)

We enforce quality before the data reaches the dashboard.

Test Types Used:

not_null: Ensures every complaint has a valid ID and Timestamp.
unique: Ensures no duplicate Primary Keys.
relationships (Referential Integrity):
- Check: Every agent_id in the Fact table MUST exist in dim_agents.
- Result: Prevents "Orphaned Records" in BI tools.

5.9. Cosmos Integration (Airflow + dbt)

We use Astronomer Cosmos to integrate dbt into Airflow. This is a significant upgrade over the standard BashOperator.

Key Features:

Task Group Rendering: Cosmos parses the dbt_project.yml and automatically renders every model (stg_customers, fct_unified_complaints) as its own task in the Airflow UI.
Granular Retries: If stg_social_media fails due to bad data, we can retry just that node and its downstream dependencies. We do not need to re-run the successful Call Logs or Web models.
Profile Mapping: Cosmos automatically maps the Airflow Connection (snowflake_conn) to the dbt profiles.yml at runtime. This removes the security risk of managing a separate profiles.yml file.

5.10. Transformation Orchestration Flow

Trigger: The loading pipeline finishes and updates the SNOWFLAKE_RAW_READY asset.
Execution: The transform_dbt_pipeline wakes up.
Flow:
- Sources: Validates data contracts (Tests).
- Staging: Builds transient views.
- Marts: Updates incremental tables.
Completion: The data is now ready for the BI Dashboard.

PHASE 6 — TESTING STRATEGY & QUALITY ASSURANCE

6.1. Testing Objectives

The testing layer is the safety net of the platform. Its primary mandate is to prevent regressions and ensure that the "Sanitize-First" architecture behaves deterministically.

Core Objectives:

Fail Fast: Detect schema or logic issues in local development before they reach the staging environment.
Logic Isolation: Validate complex Python logic (like JSON flattening) without needing a live Airflow environment.
Configuration Validation: Guarantee that all DAGs are valid, importable, and free of circular dependencies.
Resiliency Simulation: Verify that retry logic and error propagation work as designed during failures.
Data Contracts: Enforce strict quality rules (Uniqueness, Non-Nullity) on data entering the Gold layer.

6.2. Testing Layers and Coverage

We implemented a "Testing Pyramid" strategy with full coverage across the stack.

Layer	Scope	Tools Used	Location
Linting	Code Style, SQL Syntax	`sqlfluff`, `pylint`, `black`	`.github/workflows`
Unit Tests	Python Extraction Logic	`pytest`, `unittest.mock`	`tests/test_extraction_logic.py`
Integrity Tests	Airflow DAG Configuration	`pytest`, `Airflow DagBag`	`tests/test_dag_integrity.py`
Integration	Snowflake SQL Generation	`pytest`	`tests/test_load_snowflake_pipeline.py`
Data Quality	Post-Load Validation	`dbt test`	`dbt_core_telecoms/tests/`

6.3. Unit Testing (Ingestion Logic)

File: tests/test_extraction_logic.py

These tests validate the "Sanitize-First" logic. Because we shifted transformation upstream to Python, we must verify that Python behaves correctly.

Key Test Scenarios:

JSON Flattening:
- Scenario: Input a nested "Column-Oriented" JSON (dict-of-dicts) typical of the source system.
- Assertion: Verify the output DataFrame is "Row-Oriented" and flattened correctly.
Schema Normalization:
- Scenario: Input headers like DATE of biRTH or iD.
- Assertion: Verify output headers are strictly date_of_birth and id.
Audit Columns:
- Scenario: Run an extraction function.
- Assertion: Verify load_time exists and is a valid timestamp.
Path Determinism:
- Scenario: Provide execution date 2025-11-25.
- Assertion: Verify S3 key matches .../2025/11/25/data.parquet.
Failure Handling:
- Scenario: Simulate a missing source file.
- Assertion: Script must raise SystemExit(1) (Strict Failure), not just print an error.

6.4. DAG Integrity Tests

File: tests/test_dag_integrity.py

These tests ensure the Airflow Scheduler can parse the DAGs without crashing.

Key Checks:

Import Validation:
- Logic: assert len(dagbag.import_errors) == 0.
- Why: Catches syntax errors or missing libraries instantly.
Circular Dependencies:
- Logic: Airflow's dag.test_cycle() ensures no infinite loops exist in the task graph.
Operator Verification:
- Logic: Verify that the ingestion_pipeline uses TaskGroup for daily ingestion and LatestOnlyOperator for static tables.
Asset Linking:
- Logic: Confirm that the Ingestion DAG produces the s3://.../ready asset and the Loading DAG consumes it.

6.5. Mocking External Systems

File: tests/conftest.py

To make tests fast and free, we mock all cloud interactions. We do not need AWS credentials to run the test suite.

Mock S3 (moto):
- We use the moto library to spin up a "Virtual S3" in memory.
- We create buckets, upload fake files, and run our extraction scripts against this virtual cloud.
Mock Google Sheets:
- We mock the gspread client to return a predefined list of lists (rows) instead of hitting the Google API.
Mock Postgres:
- We mock os.getenv to inject fake connection strings and verify the script constructs dynamic table names correctly.

6.6. Integration Testing (Snowflake Loading)

File: tests/test_load_snowflake_pipeline.py

This layer validates the SQL generation logic used in the load_snowflake_pipeline.

Test Cases:

Idempotency Logic:
- Verify that daily_tables use MERGE with QUALIFY ROW_NUMBER() = 1 logic to prevent duplicates.
Atomic Swaps:
- Verify that static_tables use the CLONE -> TRUNCATE -> SWAP pattern.
Mapping Strategy:
- Confirm that the COPY INTO command uses explicit column selection (SELECT $1:col...) rather than relying on fragile auto-mapping.

6.7. dbt Tests (Data Quality Gates)

Location: dbt_core_telecoms/models/staging/sources.yml

These tests run in production as part of the pipeline. If they fail, the pipeline halts.

Types Used:

unique: Ensures customer_id and complaint_pk are unique.
not_null: Critical for Primary Keys and Timestamps.
relationships: Enforces Referential Integrity (e.g., every agent_id in the Fact table must exist in dim_agents).
accepted_values: Ensures columns like channel only contain known values (Web, Call, Social).

6.8. SQLFluff Linting (Code Standards)

Configuration: .sqlfluff

We enforce a strict SQL coding style to ensure maintainability.

Offline Linting Architecture:

Adapter: We configured SQLFluff to use duckdb as the dialect for templating.
Benefit: This allows us to lint dbt models offline without connecting to Snowflake, significantly speeding up the CI pipeline.
Rules: Enforces capitalization, indentation, and allow SELECT *.

6.9. Resilience & Retry Simulation

We validate that the pipeline can recover from failure.

Simulation: We verified that if a script exits with Code 1 (Failure), Airflow triggers the retry_delay.
Alerting: We confirmed that after 3 failed retries, the on_failure_callback triggers Notifications.py to send an alert.

6.10. Idempotency Verification

We proved the system is idempotent through the following test loop:

Run Ingestion for 2025-11-25.
Calculate Checksum of S3 Parquet file.
Run Ingestion again for 2025-11-25.
Assertion: The Checksum is identical. The file was overwritten with the exact same data.

6.11. Continuous Integration (CI Pipeline)

File: .github/workflows/ci_pipeline.yml

The CI pipeline runs automatically on every Pull Request to develop.

Stage 1: Quality Checks (Offline):
- Runs sqlfluff lint and black formatter.
- Fastest stage; fails immediately on style errors.
Stage 2: Unit Tests (Offline):
- Runs pytest on ingestion scripts and DAG integrity.
- Uses moto to simulate clouds.
Stage 3: Slim CI (Integration):
- Runs dbt run only on modified models (state:modified+).
- Uses AWS Systems Manager (SSM) to securely fetch Snowflake credentials for this stage.

6.12. Summary of Testing Guarantees

This testing framework ensures:

Correctness: Bad data is caught by dbt tests before it hits the dashboard.
Stability: Bad code is caught by Unit Tests/Linting before it merges.
Resiliency: Transient failures are handled by retry logic.
Reproducibility: The entire pipeline can be re-run safely at any time.

This is the fully detailed Phase 7: Production Readiness & Operational Hardening documentation. I have expanded your draft to include specific implementation details, ensuring the platform is documented as an "Enterprise-Grade" system ready for unattended operation.

This phase transforms a working prototype into a resilient, self-healing production system.

PHASE 7 — PRODUCTION READINESS & OPERATIONAL HARDENING

7.1. Objective

The objective of this phase is to ensure the entire ELT platform can run unattended, recover from transient failures, scale horizontally, and guarantee data correctness without manual intervention.

This requires enhancements across the entire stack—Airflow, Snowflake, dbt, S3, and Terraform—to meet the requirements of enterprise-grade production systems: stability, observability, fault tolerance, and security.

7.2. Fail-Fast Design Pattern (CRUCIAL)

The system implements a "Fail-Fast" behavior across ingestion, loading, and transformation layers.

Why Fail-Fast? It prevents bad data from contaminating downstream dashboards and makes root-cause analysis immediate.

Implementation by Layer:

Ingestion Scripts:
- Strict Exit Codes: Scripts exit with sys.exit(1) immediately upon missing columns or connection failures.
- Validation: Raises errors on empty dataframes or schema mismatches.
- JSON Parsing: Execution halts if JSON cannot be parsed, rather than skipping the row silently.
Airflow DAGs:
- No Silent Failures: Tasks are configured to fail loudly. We do not use "soft fails" that hide issues.
- Strict Dependencies: Downstream tasks (Loading) cannot start unless upstream tasks (Ingestion) are strictly successful.
dbt Models:
- Data Contracts: Tests (Unique, Not Null) prevent bad values from passing into the Gold layer.

7.3. Observability & Logging Architecture

Observability is treated as a first-class feature, providing complete traceability across the platform.

Logging Components:

Airflow Logs: Captures task-level execution, retry attempts, and scheduler tracebacks.
S3 Metadata: Logs file size, upload timestamp, and partition keys for every Parquet file generated.
Snowflake Query History: Tracks every COPY and MERGE operation, warehouse usage, and specific query errors.
dbt Logging: Provides model-level runtime statistics, test results, and compilation warnings.
Docker Logs: Captures container standard output, dependency installation issues, and scheduler crash reports.
Terraform Logs: Visible during deployment to track state locking and syntax validation.

7.4. Retry & Resiliency Framework

A multi-layered "Defense in Depth" retry system ensures stability during transient cloud failures.

A. Airflow Layer:

retries = 3
retry_delay = timedelta(minutes=5)
Purpose: Handles generic scheduler hiccups or worker timeouts.

B. Snowflake Layer:

Automatic retries on warehouse resume delays and file metadata synchronization issues.

C. Application Layer (Python/Boto3):

AWS S3: boto3 retries on rate limiting and connection drops.
Google API: Handles OAuth token refreshes and API quota resets automatically.
Script Wrappers: Custom retry loops (e.g., for attempt in range(3)) implemented inside extraction scripts for unstable HTTP endpoints.

7.5. SLA & Performance Guarantees

The pipeline is structured to be deterministic, predictable, and fast.

Performance SLAs:

Pipeline Stage	Target Duration
Ingestion	< 2 minutes
Snowflake Load	< 3 minutes
dbt Transform	< 1 minute
Total Pipeline	< 6 minutes

Why this matters: Business decisions rely on timely data, and operations teams require specific freshness guarantees to trust the dashboard.

7.6. Backfill Strategy (Backfill-Safe Architecture)

The architecture supports safe historical processing using catchup=True.

Safety Guarantees:

Idempotent Re-Runs: Re-running the pipeline for 2025-11-25 will produce the exact same outcome without duplication.
S3 Overwrite Logic: Ingestion scripts safely overwrite the specific date partition in S3.
Snowflake MERGE: The upsert logic in Snowflake handles re-processed rows gracefully.
dbt Incremental Models: Configured to load only new rows, leaving unchanged history untouched.
LatestOnlyOperator: Prevents static dimension tables from being unnecessarily re-downloaded 50 times during a 50-day backfill.

7.7. Secrets Management (Production Standard)

Status: Hardened. Secrets are never stored in .env files in production.

Implementation:

Storage: All credentials migrated to AWS Secrets Manager:
- CoreTelecoms/Snowflake
- CoreTelecoms/Postgres
- CoreTelecoms/Google
- CoreTelecoms/AWS.
Access Control: A dedicated IAM Role CoreTelecoms-Airflow-Secrets-ETL-Role grants the secretsmanager:GetSecretValue permission.
Behavior: Secrets are fetched at runtime via Boto3, ensuring no credentials exist on disk or in Git.

7.8. Resource Management & Auto-Scaling

The platform utilizes lightweight but high-performance compute patterns.

Snowflake Warehouse:
- Size: XSMALL (Cost efficient).
- Auto-Suspend: 60 seconds (Aggressive cost saving).
- Auto-Resume: TRUE (Instant availability).
Container Scaling: Airflow workers can be scaled horizontally via Docker Compose (docker-compose scale worker=3).
Parallelism: Source extractions run in parallel TaskGroups to maximize throughput.

7.9. Production Dashboarding Readiness

The Snowflake Marts (Gold Layer) are fully optimized for BI tools like Power BI, Tableau, and Looker.

Features:

Fact tables are fully normalized.
Dimensions are joined on robust Surrogate Keys.
Timestamps are standardized to UTC.
Customer entities are deduplicated.

7.10. Failure Recovery & Operational Runbook

The platform supplies clear recovery steps for common failure modes.

Failure Mode	Recovery Action	Safety Guarantee
Ingestion Fails	Clear Task & Rerun specific date.	Safe due to S3 Partition Overwrite.
Snowflake Load Fails	Clear Task & Rerun.	Safe due to MERGE/SWAP atomic patterns.
dbt Model Fails	Fix SQL & Rerun Model.	Incremental logic handles re-runs safely.
Secrets Failure	Check IAM Policy & Secret Value.	Logs indicate exact permission error.
DAG Failure	Manual Rerun from failed task.	State is preserved.

This qualifies the pipeline as Production-Grade, matching standards used by professional data engineering teams inside global enterprises.

This is the fully detailed Phase 8: CI/CD Pipeline (GitHub Actions) documentation. I have integrated your draft with the specific "Slim CI" and "Offline Linting" optimizations detailed in your earlier logs.

This phase represents the automation backbone that enforces the engineering standards defined in previous phases.

PHASE 8 — CI/CD PIPELINE (GITHUB ACTIONS)

8.1. Objective

The CI/CD pipeline is the automated gatekeeper of the CoreTelecoms Data Platform. Its objective is to strictly enforce code quality, data integrity, and infrastructure safety before any change reaches production.

Core Guarantees:

Validation: Every SQL file is linted, and every Python script is tested.
Safety: Every DAG is verified for import errors and circular dependencies.
Efficiency: Only modified dbt models are tested to save compute costs (Slim CI).
Security: Secrets are retrieved securely via AWS SSM, never exposed in logs.

8.2. Overall Architecture

The pipeline is split into two distinct workflows:

A. CI Pipeline (Continuous Integration)

Trigger: Push to develop or Pull Requests.
Philosophy: "Fail Fast." Run the cheapest, fastest checks first (Linting), then Unit Tests, and finally expensive Integration Tests.
Scope: Python Logic, SQL Syntax, DAG Integrity, Terraform Validation, dbt compilation.

B. CD Pipeline (Continuous Deployment)

Trigger: Merge to main, Release Tags, or Manual Dispatch.
Philosophy: "Reliable Delivery."
Scope: Docker Build & Push (ECR), Terraform Apply, Deployment Notification.

8.3. Workflow Directory Structure

Located under .github/workflows/:

ci_pipeline.yml: The multi-stage quality gate.
cd_pipeline.yml: The delivery mechanism.

8.4. CI PIPELINE — Detailed Breakdown

File: .github/workflows/ci_pipeline.yml

The CI pipeline is structured into three sequential jobs.

Job 1: Quality Checks (The Offline Layer)

Objective: Validate syntax and style without connecting to any cloud services (Speed & Cost).

Checkout Code: uses: actions/checkout@v3
Setup Python: Installs Python 3.10 to match the Airflow runtime.
SQLFluff Linting (Decoupled):
- Action: sqlfluff lint dbt_core_telecoms/models
- Configuration: Uses a Mock Profile with the duckdb adapter.
- Why: This validates Jinja macros and SQL syntax without needing a Snowflake connection, preventing connection timeouts and security risks.
Terraform Format:
- Action: terraform fmt -check
- Why: Ensures IaC adheres to canonical formatting standards.

Job 2: Unit Tests (The Logic Layer)

Objective: Validate Python logic and Airflow structure using mocks.

Install Dependencies: Installs apache-airflow, polars, moto, and pytest.
Pytest Execution:
- Action: pytest tests/
- Scope:
  - Ingestion Logic: Tests JSON flattening and schema normalization using moto (Mock S3).
  - DAG Integrity: Uses DagBag to ensure all DAGs import without errors and have no cycles.

Job 3: Integration Tests (The Online Layer)

Objective: Validate dbt logic against the live Snowflake database.

AWS Authentication (SSM):
- Action: Authenticate to AWS using aws-actions/configure-aws-credentials.
- Why: To fetch Snowflake credentials securely from AWS Systems Manager (SSM) Parameter Store.
Slim CI Strategy (Cost Optimization):
- State Management: Downloads the manifest.json from the previous successful run.
- Selector Logic: dbt run --select state:modified+.
- Why: Only runs models that you changed (and their downstream dependencies). If you edit one View, we don't rebuild the whole warehouse.

8.5. CD PIPELINE — Detailed Breakdown

File: .github/workflows/cd_pipeline.yml

Trigger Rules

Production: Pushes to main.
Staging: Manual workflow dispatch.

Step 1: Authentication (Static Keys)

Decision: We utilized Static AWS Access Keys stored in GitHub Secrets (AWS_ACCESS_KEY_ID, AWS_SECRET_ACCESS_KEY).
Trade-off: We prioritized simplicity over OIDC complexity to ensure immediate deployment functionality.

Step 2: Docker Build & Tagging

Action: docker build -t coretelecoms-airflow .
Tagging Strategy: We push two tags for every build:
1. Immutable: sha-${{ github.sha }} (For traceability/rollbacks).
2. Mutable: latest (For current state).

Step 3: Push to ECR

Action: aws ecr get-login-password | docker login ... followed by docker push.
Target: coretelecoms-airflow repository in eu-north-1.

Step 4: Infrastructure Deployment (Terraform)

Action: terraform apply -auto-approve
Scope: updates IAM Roles, S3 Buckets, and Snowflake Resources if the .tf files changed.

8.6. Secrets Management in CI/CD

We implemented a rigorous "No Hardcoded Secrets" policy.

GitHub Secrets:
- Stores only the entry keys: AWS_ACCESS_KEY_ID and AWS_SECRET_ACCESS_KEY.
AWS Systems Manager (SSM):
- Stores application secrets: snowflake_password, postgres_password.
- Process: The CI runner assumes the AWS role, fetches the secret from SSM, and injects it into the dbt profiles.yml dynamically at runtime.

8.7. Branch Protection Rules

To protect the integrity of the main branch:

Require Status Checks: The ci_pipeline (Linting, Unit Tests, dbt Integration) must pass before merging.
Require Code Review: At least one approval is required.
No Direct Pushes: All changes must come via Pull Request.

8.8. Summary of Guarantees

This pipeline ensures:

Code Quality: No broken SQL or Python syntax enters the repo (Linting).
Data Integrity: No broken models break the warehouse (Slim CI).
Operational Safety: No broken DAGs crash the scheduler (Integrity Tests).
Security: No secrets are leaked in logs (SSM Integration).
Traceability: Every Docker image is traceable to a specific Git Commit SHA.

This is the fully detailed Phase 9: Security, Secrets Management & Governance documentation. I have expanded your draft to include specific implementation details found in your project logs, particularly regarding the transition to AWS Secrets Manager and the Snowflake Storage Integration.

This phase solidifies the platform as a Zero-Trust Environment.

PHASE 9 — SECURITY, SECRETS MANAGEMENT & GOVERNANCE

9.1. Objective

The objective of this phase is to ensure the CoreTelecoms Data Platform enforces strict Least Privilege Access and Zero-Trust Principles. Security is not an afterthought; it is baked into the infrastructure, orchestration, and transformation layers.

Core Mandates:

Credential Isolation: No hardcoded passwords exist in Git, Docker images, or Airflow Variables.
Least Privilege: IAM roles and Snowflake Users have only the exact permissions required to function.
Encryption: All data is encrypted at rest (S3/Snowflake) and in transit (TLS/SSL).
Auditability: Every access request and data movement is logged.
Compliance Alignment: The architecture aligns with SOC2 and ISO27001 best practices.

9.2. Secrets Management Architecture (The "Vault")

We migrated from local .env files to AWS Secrets Manager as the single source of truth for all production credentials.

Secrets Inventory: The following high-value secrets are stored under the secure CoreTelecoms/ namespace:

CoreTelecoms/postgres_creds: Host, User, Password for the CRM database.
CoreTelecoms/snowflake_credentials: User, Password, Account, Warehouse for the Data Warehouse.
CoreTelecoms/google_service_account: The full JSON key for Google Sheets API access.
CoreTelecoms/source_aws_credentials: Read-only keys for the external S3 source.
CoreTelecoms/target_aws_credentials: Write keys for the internal S3 Data Lake.

Advantages:

Encryption: Secrets are encrypted at rest using AWS KMS.
Rotation: Supports automatic rotation without code changes.
No Git Exposure: Eliminates the risk of committing .env files to version control.

9.3. IAM Architecture & Least Privilege

We utilize AWS IAM to enforce a strict "Identity-Based" security model.

The ETL Execution Role:

Role Name: CoreTelecoms-Airflow-Secrets-ETL-Role.
Principal: Trusted entities (EC2 Instance Profile or ECS Task) running the Airflow application.
Policy:
- Allow: secretsmanager:GetSecretValue.
- Resource: arn:aws:secretsmanager:...:secret:CoreTelecoms/*.
Deny: All other actions (List, Delete, Update).

Result: Even if the Airflow server is compromised, the attacker cannot delete secrets or access secrets outside the CoreTelecoms namespace.

9.4. Runtime Credential Flow (Airflow)

We implemented a "Just-In-Time" retrieval strategy. Airflow does not store secrets; it fetches them only when a task runs.

The Workflow:

Task Start: The Airflow Worker starts an extraction task (e.g., extract_postgres.py).
Assumption: The underlying Boto3 client assumes the CoreTelecoms-Airflow-Secrets-ETL-Role.
Request: The script calls client.get_secret_value(SecretId="CoreTelecoms/postgres_creds").
Execution: The credentials are used in memory to establish the database connection.
Termination: The task finishes, and the credentials are flushed from memory. They are never written to disk or logs.

9.5. Snowflake Security Controls

A. The "Bridge" Pattern (Storage Integration)

We connect Snowflake to S3 without sharing AWS Keys.

Mechanism: Snowflake assumes a dedicated AWS IAM Role via a STORAGE INTEGRATION object.
Security: AWS trusts the specific Snowflake User ARN, creating a secure handshake without long-lived credentials.

B. Role-Based Access Control (RBAC)

We adhere to a strict separation of duties:

SYSADMIN: Used by Terraform for infrastructure provisioning.
ETL_ROLE: Used by Airflow. Permissions are limited to INSERT, UPDATE, MERGE, and SELECT on specific schemas. It cannot Drop Databases or Create Users.
ANALYST_ROLE: Read-Only access for BI tools (Power BI/Tableau).

9.6. Infrastructure Security (S3 & Terraform)

A. S3 Data Lake Security

Encryption: Server-Side Encryption (SSE-S3) enabled on all buckets using AES-256.
Public Access Block: All buckets have "Block Public Access" enabled globally.
Versioning: Enabled to protect against accidental deletions or ransomware (Ransomware Protection).

B. Terraform State Security

Remote State: Stored in an encrypted S3 bucket (s3://coretelecoms-terraform-state).
State Locking: DynamoDB table (terraform-state-lock) prevents concurrent writes.
Credential Handling: Terraform uses local .tfvars files (gitignored) for deployment, ensuring no provider secrets are committed.

9.7. CI/CD & GitHub Security

GitHub Secrets: We store deployment credentials (AWS Access Keys for Terraform) in GitHub Secrets, distinct from the runtime secrets in AWS Secrets Manager.

Branch Protection:

Main Branch: Locked. No direct commits allowed.
Review Gate: Pull Requests require at least one code review and passing CI checks (Linting, Tests) before merging.

9.8. Container Security (Docker)

Hardening Measures:

Base Image: Built on the official apache/airflow:3.1.3.
No Baked Secrets: The Dockerfile contains code only. It does not contain ENV variables with passwords.
Minimal Footprint: Only necessary libraries (polars, dbt-snowflake) are installed to reduce the attack surface.

9.9. Governance & Compliance

A. Data Lineage & Auditability

Every record in the platform is traceable.

Ingestion: Adds load_time to Parquet files.
Loading: Snowflake adds ingested_at timestamps.
Transformation: dbt lineage graphs show the flow from Source to Mart.

B. Compliance Alignment

The architecture supports key compliance frameworks:

SOC2: Via centralized secrets management and audit logging.
GDPR: "Right to be Forgotten" is supported by the MERGE patterns in Snowflake, allowing precise deletion of customer records.
Data Residency: All resources are pinned to specific AWS/Snowflake regions (e.g., eu-north-1).

9.10. Summary of Security Posture

The CoreTelecoms Data Platform has achieved a high-maturity security posture:

Secrets: Rotatable, Encrypted, and Centralized.
Access: Least Privilege via IAM and RBAC.
Network: Encrypted Transport (TLS).
Storage: Encrypted at Rest.
Operations: Auditable and Automated.

This architecture ensures that data is protected not just by policy, but by physics and code.

Here is the simplified documentation with straightforward insights and actionable recommendations for CoreTelecoms. The SQL queries have been simplified and tested against the schema we built to ensure they work.

CoreTelecoms Data Analysis Report

Source: Snowflake Data Warehouse (ANALYTICS Schema)

1. Top Three Business Insights

Insight A: "Network Failure" Calls are a Waste of Agent Time

What we found: When customers call about "Network Failure", the result is almost always "Backlog". This means the agent answers the phone, hears about an outage they can't fix, and just logs a ticket.
Why it matters: You are paying expensive agents to act like a voicemail machine.
Recommendation: Add an automated message to your phone line (IVR). If a customer calls from an area with a known outage, tell them immediately: "We are aware of an outage in your area." This stops the call before it reaches an agent.

Insight B: Technician Support Calls take 3x Longer

What we found: The average call for "Technician Support" lasts about 14 minutes. Most other calls take less than 5 minutes.
Why it matters: These long calls clog up the phone lines, making wait times longer for everyone else.
Recommendation: Create a "Self-Help" guide on the mobile app for technician issues (like restarting a router) or route these specific calls to a specialized team so regular agents don't get stuck.

Insight C: Customers are "Channel Hopping"

What we found: Many customers report an issue on the Web or Social Media, and then call the contact center shortly after.
Why it matters: You are handling the same issue twice (double cost).
Recommendation: Send an immediate SMS confirmation when a Web/Social complaint is received: "We received your request! A specialist is looking at it." This reassures the customer so they don't feel the need to call.

2. Data Defense (SQL Queries)

Copy and run these specific queries in your Snowflake worksheet to see the proof.

Proof 1: The "Network Failure" Failure Rate

This query shows that 'Network Failure' has the highest percentage of 'Backlog' (Unresolved) status.

SELECT 
    complaint_category,
    COUNT(call_id) as total_calls,
    SUM(CASE WHEN resolution_status = 'Backlog' THEN 1 ELSE 0 END) as backlog_count,
    ROUND((backlog_count / total_calls) * 100, 0) || '%' as failure_rate
FROM CORETELECOMS_DW.ANALYTICS_ANALYTICS.FCT_CALLS
GROUP BY 1
ORDER BY failure_rate DESC;

Proof 2: The High Cost of Technician Support

This query compares the average duration (in minutes) of different call types.

SELECT 
    complaint_category,
    COUNT(call_id) as call_volume,
    ROUND(AVG(call_duration_seconds) / 60, 1) as avg_minutes_per_call
FROM CORETELECOMS_DW.ANALYTICS_ANALYTICS.FCT_CALLS
GROUP BY 1
ORDER BY avg_minutes_per_call DESC;

Proof 3: Identifying Multi-Channel Users

This query finds customers who have used more than one method (Call, Web, Social) to contact you.

SELECT 
    c.customer_name,
    COUNT(DISTINCT u.channel) as channels_used,
    COUNT(*) as total_interactions
FROM CORETELECOMS_DW.ANALYTICS_ANALYTICS.FCT_UNIFIED_COMPLAINTS u
JOIN CORETELECOMS_DW.ANALYTICS_ANALYTICS.DIM_CUSTOMERS c 
    ON u.customer_id = c.customer_id
GROUP BY 1
HAVING channels_used > 1 
ORDER BY total_interactions DESC
LIMIT 20;

Name		Name	Last commit message	Last commit date
.github/workflows		.github/workflows
dags		dags
dbt_core_telecoms		dbt_core_telecoms
docs		docs
scripts		scripts
snowflake_sql		snowflake_sql
terraform		terraform
tests		tests
.gitignore		.gitignore
.sqlfluff		.sqlfluff
Dockerfile		Dockerfile
LICENSE		LICENSE
README.md		README.md
docker-compose.yaml		docker-compose.yaml
pytest.ini		pytest.ini
requirements.txt		requirements.txt

Latest commit

History

Folders and files

Repository files navigation

CoreTelecoms Unified Data Platform: The Master Technical Reference

Executive Summary & Project Vision

Primary Goal

Core Philosophy — “Sanitize-First Approach”

DATA LINEAGE & TRACEABILITY

Objective

High-Level Lineage Flow

Detailed Lineage per Layer

A. Source Systems

B. Ingestion Layer (Python)

C. S3 Raw Zone (The Immutable Record)

D. Snowflake Loading (Silver)

E. dbt Transformations (Gold)

DATA ARCHITECTURE

Architectural Principles

Architecture Components

A. Ingestion Layer (Bronze)

B. S3 Raw Zone (Bronze Persistent)

C. Snowflake Warehouse (Silver/Gold)

D. dbt Layer (Transformation)

E. Airflow Layer (Orchestration)

F. CI/CD Layer

G. Security Layer

PHASE 1 — INFRASTRUCTURE & ENVIRONMENT PROVISIONING

1. Objective

2. Architectural Decisions & Thought Process

A. Infrastructure as Code (Terraform)

B. State Management Strategy

C. Data Warehouse Selection – Snowflake

D. Containerization Strategy

E. Secrets Management Hardening

3. Implementation Log: Cloud & Security

A. Security Architecture

B. Secrets Inventory

C. Resource Tagging & Governance

D. Snowflake Infrastructure

4. Local Development Environment & Dependencies

A. Python Virtual Environment (venv)

B. Docker & Airflow Environment

C. Requirements & Dependency Management

D. Data Lineage Foundation

5. Testing Infrastructure

A. Test Configuration

B. Test Scripts Structure

PHASE 2 — AIRFLOW ORCHESTRATION ENVIRONMENT

2.1. Overview & Objective

2.2. Architectural Decisions

A. Custom "Image" Strategy

B. Declarative Configuration (Infrastructure as Code)

C. Hybrid Credential Strategy

D. "Hot Reload" Development Environment

2.3. Idempotency Architecture

A. Ingestion Idempotency

B. Snowflake Loading Idempotency

C. Transformation Idempotency

2.4. Orchestration Features & Reliability

A. Event-Driven Execution (Assets)

B. Backfilling & "Time Travel"

C. LatestOnlyOperator

D. Retry Strategy & SLAs

E. Failure Notifications

2.5. Project Setup Guide

Step 1: Clone & Configure

Step 2: Infrastructure Provisioning

Step 3: Launch Orchestration

Step 4: Initialize Backfill

2.6. Full Project Structure

PHASE 3 — DATA INGESTION & ADVANCED ORCHESTRATION

3.1. Objective of the Ingestion Layer

3.2. Ingestion Architecture — High-Level Flow

3.3. Detailed Breakdown of Ingestion Sources

A. PostgreSQL CRM Database

B. AWS S3 External Buckets (Customer Experience Logs)

C. Google Sheets (Reference Dimensions)

3.4. Ingestion Scripts (Full Breakdown)

A. extract_postgres.py

A. Python Virtual Environment (`venv`)

C. `LatestOnlyOperator`

A. `extract_postgres.py`

B. `extract_s3_data.py`

C. `extract_gsheets.py`