Write DBT code to ETL code from data warehouse bronze layer to gold layer
Step 2: Dimensional Model for Sale Processes
Overview
Fact Table
Dim Date
Dim Territory
Dim Product
fact_sale
x
x
x
Original table ERD Structure
Table Explanations
Fact Table
Table Name
Purpose
Description
Key Metrics
Grain
fact_sale
Central fact table for sales transactions
Contains measurable sales events with foreign keys to dimension tables. Each row represents a sales order line item with associated quantities, amounts, and dimensional context.
Update profiles.yml file (usually located in ~/.dbt/profiles.yml):
Step 4: Install Custom Data Suite Adapter (Optional)
Step 5: Configure Project Structure
Update dbt_project.yml:
Step 6: Create Layer Structure
Step 7: Define Data Sources
Create models/staging/sources.yml:
Step 8: Create Staging Models
Create staging models to clean and standardize bronze layer data:
Step 9: Build Dimensional Models
Create dimensional model files:
models/marts/core/dim_date.sql
models/marts/core/dim_territory.sql
models/marts/core/dim_product.sql
models/marts/core/fact_sale.sql
Step 10: Run DBT Pipeline
Step 11: Schedule with Airflow (Optional)
Create DAG for automated runs:
Step 12: Data Quality & Monitoring
Set up data quality tests in tests/ directory
Configure alerting for test failures
Monitor pipeline performance and data freshness
Set up incremental model strategies for large tables
Create and deploy code ETL to Airflow (production)
Overview
This section covers deploying the complete AdventureWorks ETL pipeline to Apache Airflow for production use, including LogStash data ingestion, DBT transformations, and monitoring.
Production Architecture
Prerequisites
Docker and Docker Compose installed
Apache Airflow 2.7+ running
ClickHouse database accessible
DBT project configured (from previous section)
LogStash configuration files ready
Step 1: Airflow Environment Setup
1.1 Docker Compose Configuration
Create docker-compose.yml for Airflow production setup:
1.2 Install Required Packages
Create requirements.txt for additional Python packages:
Build custom Airflow image with requirements:
Step 2: Create Production DAGs
2.1 Main ETL Pipeline DAG
Create dags/adventureworks_etl_pipeline.py:
2.2 Incremental Loading DAG
Create dags/adventureworks_incremental_etl.py:
Step 3: Configuration Management
3.1 Airflow Variables
Set up Airflow variables via Web UI or CLI:
3.2 Connections
Create Airflow connections for database access:
Step 4: Deployment Process
4.1 Pre-deployment Checklist
4.2 Production Deployment
Step 5: Monitoring and Alerting
5.1 Set up Airflow Monitoring
5.2 Alerting Configuration
Add to airflow.cfg:
Step 6: Performance Optimization
6.1 Resource Allocation
6.2 Parallel Execution
Step 7: Backup and Recovery
7.1 Automated Backups
7.2 Disaster Recovery Plan
Data Recovery: Restore from ClickHouse backups
Pipeline Recovery: Redeploy from version control
State Recovery: Restore Airflow metadata database
Validation: Run data quality checks post-recovery
Step 8: Security Best Practices
Secrets Management: Use Airflow's secret backend
Access Control: Implement RBAC for Airflow users
Network Security: Use VPN/private networks
Encryption: Enable SSL/TLS for all connections
Audit Logging: Monitor all pipeline activities
Step 9: Documentation and Handover
Runbook: Document troubleshooting procedures
Architecture Diagrams: Maintain up-to-date system diagrams
Change Management: Document all configuration changes
Training: Provide team training on pipeline operations
# Create network for container communication
$ docker network create datasuite-network
# Run MySQL with persistent storage
$ docker run --name mysql \
--network datasuite-network \
-e MYSQL_ROOT_PASSWORD=password \
-e MYSQL_DATABASE=adventureworks \
-p 3306:3306 \
-v mysql-data:/var/lib/mysql \
-d mysql:8.0
# Load Adventure Works 2019 data from local file
$ docker exec -i mysql mysql -uroot -ppassword adventureworks < docs/data/AdventureWorks2019.sql
# Verify data was loaded successfully
$ docker exec mysql mysql -uroot -ppassword adventureworks -e "SHOW TABLES;"
$ docker exec mysql mysql -uroot -ppassword adventureworks -e "SELECT COUNT(*) as table_count FROM information_schema.tables WHERE table_schema='adventureworks';"
class Sales_SalesOrderDetail
class Sales_SaleOrderHeader
class Sales_Customer
class Sales_SalesTerritory
Sales_SaleOrderHeader --> Sales_SalesOrderDetail
Sales_SaleOrderHeader -> Sales_Customer
Sales_SaleOrderHeader -> Sales_SalesTerritory
$ docker run --name clickhouse \
--network datasuite-network \
-p 8123:8123 \
-p 9000:9000 \
-e CLICKHOUSE_USER=admin \
-e CLICKHOUSE_PASSWORD=clickhouse123 \
-e CLICKHOUSE_DEFAULT_ACCESS_MANAGEMENT=1 \
-v clickhouse-data:/var/lib/clickhouse \
-d clickhouse/clickhouse-server:latest
# Verify ClickHouse is running with authentication
$ curl -u admin:clickhouse123 http://localhost:8123/ping
# Test connection and create database
$ curl -u admin:clickhouse123 http://localhost:8123/ -d "CREATE DATABASE IF NOT EXISTS bronze_layer"
$ curl -u admin:clickhouse123 http://localhost:8123/ -d "CREATE DATABASE IF NOT EXISTS gold_layer"
# Create directory for MySQL connector
$ mkdir -p logstash-drivers
# Download MySQL JDBC driver using Docker container
$ docker run --rm -v $(pwd)/logstash-drivers:/drivers alpine:latest \
sh -c "apk add --no-cache wget && \
wget -O /drivers/mysql-connector-java.jar \
https://repo1.maven.org/maven2/mysql/mysql-connector-java/8.0.9-rc/mysql-connector-java-8.0.9-rc.jar"
# Run LogStash with driver mounted
$ docker run --name logstash \
--network datasuite-network \
-p 5044:5044 \
-v $(pwd)/logstash.conf:/usr/share/logstash/pipeline/logstash.conf \
-v $(pwd)/logstash-drivers:/usr/share/logstash/drivers \
-d docker.elastic.co/logstash/logstash:8.11.0
version: 2
sources:
- name: bronze_layer
description: "Raw data from LogStash ETL pipeline"
tables:
- name: sales_orders
description: "Sales order header information"
columns:
- name: salesorderid
description: "Primary key for sales order"
tests:
- unique
- not_null
- name: sales_order_details
description: "Sales order line item details"
- name: customers
description: "Customer master data"
- name: products
description: "Product catalog information"
-- models/staging/stg_sales_orders.sql
{{ config(materialized='view') }}
SELECT
salesorderid,
customerid,
territoryid,
orderdate,
duedate,
shipdate,
status,
subtotal,
taxamt,
freight,
totaldue,
modifieddate
FROM {{ source('bronze_layer', 'sales_orders') }}
WHERE salesorderid IS NOT NULL
# Test connections
dbt debug
# Run staging models
dbt run --models staging
# Run dimensional models
dbt run --models marts
# Run all tests
dbt test
# Generate documentation
dbt docs generate
dbt docs serve
from airflow import DAG
from airflow.operators.bash_operator import BashOperator
from datetime import datetime, timedelta
dag = DAG(
'adventureworks_dbt_pipeline',
default_args={'retries': 1},
schedule_interval='0 2 * * *', # Daily at 2 AM
start_date=datetime(2025, 1, 1)
)
dbt_run = BashOperator(
task_id='dbt_run',
bash_command='cd /path/to/adventureworks_analytics && dbt run',
dag=dag
)
dbt_test = BashOperator(
task_id='dbt_test',
bash_command='cd /path/to/adventureworks_analytics && dbt test',
dag=dag
)
dbt_run >> dbt_test
# Make deploy script executable
chmod +x deploy.sh
# Run pre-deployment validation
./deploy.sh
# Deploy to production
docker-compose up -d
# Verify deployment
docker-compose ps
docker-compose logs airflow-scheduler
# Enable DAGs
airflow dags unpause adventureworks_etl_pipeline
airflow dags unpause adventureworks_incremental_etl
# Create monitoring DAG: dags/pipeline_monitoring.py
from airflow import DAG
from airflow.operators.python_operator import PythonOperator
from datetime import datetime, timedelta
def check_pipeline_health():
from airflow.models import DagRun
from datetime import datetime, timedelta
# Check if main pipeline ran successfully in last 24 hours
recent_runs = DagRun.find(
dag_id='adventureworks_etl_pipeline',
execution_start_date=datetime.now() - timedelta(days=1)
)
successful_runs = [run for run in recent_runs if run.state == 'success']
if not successful_runs:
raise Exception("No successful pipeline runs in the last 24 hours!")
return f"✅ Pipeline health OK. Last successful run: {successful_runs[-1].execution_date}"
monitoring_dag = DAG(
'pipeline_health_monitoring',
default_args={'start_date': datetime(2025, 1, 1)},
schedule_interval=timedelta(hours=6),
catchup=False
)
health_check = PythonOperator(
task_id='check_pipeline_health',
python_callable=check_pipeline_health,
dag=monitoring_dag
)
