Daily Customer Analytics Apache Airflow Project Template

A complete, production-ready Apache Airflow project template designed for daily customer data extraction, S3 staging, Snowflake loading, and Slack failure alerting.

---

Project Layout

.
├── dags/
│   └── customer_analytics_dag.py   # Airflow DAG (Postgres -> S3 -> Snowflake + Slack Alerting)
├── plugins/                         # Custom Airflow plugins & hooks
├── config/                          # Configuration adjustments
├── logs/                            # Local task execution logs
├── requirements.txt                 # Airflow provider packages (AWS, Snowflake, Postgres, Slack)
├── docker-compose.yaml              # Local multi-container setup (Webserver, Scheduler, Postgres)
└── README.md                        # Project Setup, Airflow Connections & Snowflake Guide

---

1. Quickstart: Launching the Stack

Step 1: Clone / Create Directory Structure

mkdir -p dags plugins config logs

Step 2: Set Airflow User ID (Linux/macOS)

echo "AIRFLOW_UID=$(id -u)" > .env

Step 3: Start Docker Compose Network

docker compose up -d

This command automatically:

• Starts PostgreSQL metadata database.
• Runs airflow-init to execute database migrations (airflow db migrate) and create the default admin user.
• Launches airflow-webserver on http://localhost:8080 (Username: airflow, Password: airflow).
• Launches airflow-scheduler.

---

2. Airflow UI Connection Mapping Guide

Navigate to Admin -> Connections -> Add a new record in the Airflow UI (http://localhost:8080) to configure the following 4 connections:

Connection 1: my_postgres_conn

• Conn Id: my_postgres_conn
• Conn Type: Postgres
• Host: your-operational-postgres-host (or postgres if testing locally)
• Database: production
• Login: postgres_user
• Password: postgres_password
• Port: 5432

---

Connection 2: my_aws_conn

• Conn Id: my_aws_conn
• Conn Type: Amazon Web Services
• AWS Access Key ID: YOUR_AWS_ACCESS_KEY_ID
• AWS Secret Access Key: YOUR_AWS_SECRET_ACCESS_KEY
• Extra (JSON):

{
  "region_name": "us-east-1"
}

---

Connection 3: my_snowflake_conn

• Conn Id: my_snowflake_conn
• Conn Type: Snowflake
• Host: account_identifier.snowflakecomputing.com (e.g., xy12345.us-east-1)
• Schema: ANALYTICS
• Login: AIRFLOW_USER
• Password: AIRFLOW_PASSWORD
• Account: xy12345.us-east-1 (or your Snowflake Account Identifier)
• Database: CUSTOMER_DB
• Warehouse: COMPUTE_WH
• Role: ANALYTICS_ROLE

---

Connection 4: slack_conn

• Conn Id: slack_conn
• Webhook URL / Host: https://hooks.slack.com/services/YOUR_WORKSPACE_ID/YOUR_INTEGRATION_ID/YOUR_TOKEN

---

3. Snowflake Storage Integration & Stage Setup Guide

Run the following SQL commands in Snowflake as ACCOUNTADMIN to enable secure, keyless access from Snowflake to your S3 bucket:

-- 1. Create Storage Integration in Snowflake
CREATE OR REPLACE STORAGE INTEGRATION s3_customer_analytics_int
  TYPE = EXTERNAL_STAGE
  STORAGE_PROVIDER = 'S3'
  ENABLED = TRUE
  STORAGE_AWS_ROLE_ARN = 'arn:aws:iam::123456789012:role/SnowflakeS3ReaderRole'
  STORAGE_ALLOWED_LOCATIONS = ('s3://my-company-analytics-bucket/raw/customers/');

-- 2. Describe Integration to retrieve AWS IAM User ARN and External ID
DESCRIBE STORAGE INTEGRATION s3_customer_analytics_int;

-- Copy 'STORAGE_AWS_IAM_USER_ARN' and 'STORAGE_AWS_EXTERNAL_ID' from the output 
-- and update the Trust Relationship in your AWS IAM Role (SnowflakeS3ReaderRole).

-- 3. Create Target Database, Schema, and Stage
CREATE DATABASE IF NOT EXISTS CUSTOMER_DB;
USE DATABASE CUSTOMER_DB;

CREATE SCHEMA IF NOT EXISTS ANALYTICS;
USE SCHEMA ANALYTICS;

-- Target staging table definition
CREATE TABLE IF NOT EXISTS ANALYTICS.STG_CUSTOMERS (
    customer_id INT,
    email VARCHAR(255),
    country VARCHAR(100),
    loyalty_points INT,
    loaded_at TIMESTAMP_NTZ DEFAULT CURRENT_TIMESTAMP()
);

-- External Stage referencing the Storage Integration
CREATE OR REPLACE STAGE ANALYTICS.MY_S3_STAGE
  STORAGE_INTEGRATION = s3_customer_analytics_int
  URL = 's3://my-company-analytics-bucket/raw/customers/'
  FILE_FORMAT = (
    TYPE = 'CSV'
    FIELD_DELIMITER = ','
    SKIP_HEADER = 1
    FIELD_OPTIONALLY_ENCLOSED_BY = '"'
  );

-- 4. Grant privileges to Airflow Snowflake Role
GRANT USAGE ON INTEGRATION s3_customer_analytics_int TO ROLE ANALYTICS_ROLE;
GRANT USAGE ON STAGE ANALYTICS.MY_S3_STAGE TO ROLE ANALYTICS_ROLE;
GRANT ALL ON TABLE ANALYTICS.STG_CUSTOMERS TO ROLE ANALYTICS_ROLE;

---

4. Verification & Testing

1. Trigger customer_analytics_pipeline manually in the Airflow UI.
2. Verify S3 file generation at s3://my-company-analytics-bucket/raw/customers/ds=YYYY-MM-DD/customers.csv.
3. Check target Snowflake records: SELECT * FROM CUSTOMER_DB.ANALYTICS.STG_CUSTOMERS;.
4. Test failure alerts by providing invalid query credentials to confirm Slack notification delivery.

---

5. Deploying to Kubernetes (Celery Executor)

For scaling workloads in a cluster, the project is configured to run on Kubernetes using the CeleryExecutor (managed via the official Helm chart). All deployment files are located in the kubernetes/ directory.

Quickstart Kubernetes Deployment

1. Ensure your Kubernetes cluster (e.g., Minikube) and Docker daemon are running.
2. Run the deployment script to compile the custom image and install/upgrade the Helm release:

   ./kubernetes/deploy.sh

3. If running locally on Minikube, load the custom Docker image into the Minikube registry:

   minikube image load my-airflow-app:latest

Port-Forwarding Dashboards

• Airflow Web UI:

  kubectl port-forward svc/airflow-celery-webserver 8080:8080 -n airflow

  Open http://localhost:8080 (Default login: admin / admin).
• Flower Dashboard (Celery Queue Monitor):

  kubectl port-forward svc/airflow-celery-flower 5555:5555 -n airflow

  Open http://localhost:5555 to view tasks and active worker queues.

For further instructions on secure connection secrets setup and production hardening, see the Kubernetes Deployment Guide.