Building a Production-Ready Apache Airflow Data Pipeline: PostgreSQL to Snowflake via Amazon S3

Modern data engineering pipelines must be robust, reliable, scalable, and highly observable. A common pattern in analytics is extracting transactional data from an operational database (such as PostgreSQL), staging it securely in a cloud object store (such as Amazon S3), loading it into a cloud data warehouse (such as Snowflake) for analytics, and alerting the engineering team instantly if any step fails.

To showcase these best practices, I created the aitflow project—a complete, production-ready Apache Airflow template implementing this exact ELT pipeline. It is built to support catchup-free daily schedules, date-partitioned storage paths, secure keyless Snowflake access, detailed Slack alerting, and multi-node scaling on Kubernetes via the CeleryExecutor.

In this post, we’ll dive into the architecture, examine the DAG logic, configure the connections, and detail the setup for local development and Kubernetes deployment.

🏗️ 1. Pipeline Architecture

Below is the design of the data pipeline, showcasing the integration between the source system, AWS staging, the Snowflake analytical database, and the Airflow orchestration layer:

graph TD
    subgraph Airflow [Apache Airflow Orchestration]
        DAG["customer_analytics_pipeline<br/>(Daily Execution)"]
        T1["1. PostgresToS3Operator<br/>(Extract customer rows)"]
        T2["2. S3ToSnowflakeOperator<br/>(Stage and copy into Snowflake)"]
        Callback["send_slack_failure_notification<br/>(SlackWebhookHook)"]
        
        DAG --> T1
        T1 --> T2
        T1 -.->|On Failure| Callback
        T2 -.->|On Failure| Callback
    end

    subgraph Data Sources & Staging [AWS Cloud]
        Postgres[("Operational Postgres DB<br/>production.customers")]
        S3Bucket[("Amazon S3 Bucket<br/>my-company-analytics-bucket")]
    end

    subgraph Analytics Data Warehouse [Snowflake Cloud]
        Snowflake[("Snowflake DB<br/>CUSTOMER_DB.ANALYTICS.STG_CUSTOMERS")]
        StorageIntegration["Storage Integration<br/>s3_customer_analytics_int"]
        Stage["External Stage<br/>MY_S3_STAGE"]
    end

    subgraph Alerting
        Slack["Slack Channel<br/>#alerts"]
    end

    T1 -->|Query last_active_date = ''| Postgres
    T1 -->|Upload raw/customers/ds=/customers.csv| S3Bucket
    T2 -->|Trigger Stage Load| Stage
    Stage -->|Read data using Integration| S3Bucket
    Stage -->|Insert records| Snowflake
    StorageIntegration -->|AWS IAM Role Access| S3Bucket
    Callback -->|HTTP POST Block Kit| Slack

The Pipeline Workflow:

1. Incremental PostgreSQL Extract: The PostgresToS3Operator executes a query selecting customers active on the current execution date (``) and streams the resulting CSV directly to S3.
2. Date-Partitioned Staging: Files are stored in Amazon S3 using the prefix raw/customers/ds=/customers.csv to ensure data organization, isolation, and idempotence.
3. Snowflake External Stage Copy: The S3ToSnowflakeOperator triggers a COPY INTO command loading the files into Snowflake’s target staging table. Access to S3 is granted securely via a Snowflake Storage Integration, which links Snowflake directly to an AWS IAM Role without exposing keys.
4. Observable Failure Notifications: If any task fails, the on_failure_callback constructs a Slack Block Kit alert featuring direct links to the failing task execution log for quick troubleshooting.

📁 2. Project Layout

The repository is structured logically to separate orchestration code, deployment configurations, dependencies, and environment setup:

.
├── 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

💻 3. The Airflow DAG Definition

Here is the underlying DAG definition (dags/customer_analytics_dag.py). Note the usage of dynamic Jinja templating (``) for incremental filtering and partition isolation, and the implementation of a reusable Slack Block Kit notifier:

from datetime import datetime, timedelta
import logging
from typing import Any, Dict

from airflow.decorators import dag
from airflow.models import Variable
from airflow.providers.amazon.aws.transfers.postgres_to_s3 import PostgresToS3Operator
from airflow.providers.snowflake.transfers.s3_to_snowflake import S3ToSnowflakeOperator
from airflow.providers.slack.hooks.slack_webhook import SlackWebhookHook

# Connection IDs and bucket configuration
POSTGRES_CONN_ID = "my_postgres_conn"
AWS_CONN_ID = "my_aws_conn"
SNOWFLAKE_CONN_ID = "my_snowflake_conn"
SLACK_CONN_ID = "slack_conn"

S3_BUCKET_NAME = Variable.get("CUSTOMER_ANALYTICS_S3_BUCKET", default_var="my-company-analytics-bucket")
SNOWFLAKE_STAGE = "MY_S3_STAGE"
SNOWFLAKE_SCHEMA = "ANALYTICS"
SNOWFLAKE_TABLE = "STG_CUSTOMERS"


def send_slack_failure_notification(context: Dict[str, Any]) -> None:
    """
    Failure callback function. Sends a rich Slack Block Kit notification
    containing DAG name, failing task ID, execution date, error details,
    and a direct log link button.
    """
    ti = context.get("task_instance")
    dag_id = ti.dag_id if ti else context.get("dag").dag_id
    task_id = ti.task_id if ti else "Unknown Task"
    execution_date = context.get("ds", "N/A")
    log_url = ti.log_url if ti else "#"
    exception = str(context.get("exception", "No exception message captured."))

    slack_blocks = [
        {
            "type": "header",
            "text": {
                "type": "plain_text",
                "text": "🚨 Airflow DAG Task Execution Failed",
                "emoji": True,
            },
        },
        {
            "type": "section",
            "fields": [
                {"type": "mrkdwn", "text": f"*DAG Name:*\n`{dag_id}`"},
                {"type": "mrkdwn", "text": f"*Failing Task:*\n`{task_id}`"},
                {"type": "mrkdwn", "text": f"*Execution Date:*\n`{execution_date}`"},
                {"type": "mrkdwn", "text": f"*State:*\n`FAILED`"},
            ],
        },
        {
            "type": "section",
            "text": {
                "type": "mrkdwn",
                "text": f"*Error Summary:*\n```{exception[:500]}```",
            },
        },
        {"type": "divider"},
        {
            "type": "actions",
            "elements": [
                {
                    "type": "button",
                    "text": {
                        "type": "plain_text",
                        "text": "🔍 View Task Failure Logs",
                        "emoji": True,
                    },
                    "url": log_url,
                    "style": "danger",
                }
            ],
        },
    ]

    try:
        slack_hook = SlackWebhookHook(
            slack_webhook_conn_id=SLACK_CONN_ID,
            blocks=slack_blocks,
            message=f"Task {task_id} failed in DAG {dag_id}",
        )
        slack_hook.execute()
        logging.info("Slack failure notification sent successfully.")
    except Exception as err:
        logging.error("Failed to dispatch Slack notification: %s", err)


default_args = {
    "owner": "data_platform_team",
    "depends_on_past": False,
    "email_on_failure": False,
    "email_on_retry": False,
    "retries": 1,
    "retry_delay": timedelta(minutes=5),
    "on_failure_callback": send_slack_failure_notification,
}


@dag(
    dag_id="customer_analytics_pipeline",
    default_args=default_args,
    description="Daily extraction of customer records from Postgres to S3 and Snowflake",
    schedule_interval="@daily",
    start_date=datetime(2026, 1, 1),
    catchup=False,
    max_active_runs=1,
    tags=["analytics", "customers", "postgres", "s3", "snowflake"],
)
def customer_analytics_dag():

    # 1. Extract customer rows from PostgreSQL to Amazon S3
    extract_postgres_to_s3 = PostgresToS3Operator(
        task_id="extract_postgres_to_s3",
        postgres_conn_id=POSTGRES_CONN_ID,
        aws_conn_id=AWS_CONN_ID,
        query="""
            SELECT customer_id, email, country, loyalty_points
            FROM production.customers
            WHERE last_active_date = '';
        """,
        s3_bucket=S3_BUCKET_NAME,
        s3_key="raw/customers/ds=/customers.csv",
        replace=True,
        pd_kwargs={"index": False},
    )

    # 2. Stage and load CSV data from S3 into Snowflake
    load_s3_to_snowflake = S3ToSnowflakeOperator(
        task_id="load_s3_to_snowflake",
        snowflake_conn_id=SNOWFLAKE_CONN_ID,
        s3_keys=["raw/customers/ds=/customers.csv"],
        table=SNOWFLAKE_TABLE,
        schema=SNOWFLAKE_SCHEMA,
        stage=SNOWFLAKE_STAGE,
        file_format="(TYPE = CSV FIELD_DELIMITER = ',' SKIP_HEADER = 1 FIELD_OPTIONALLY_ENCLOSED_BY = '\"')",
    )

    # Define task dependencies
    extract_postgres_to_s3 >> load_s3_to_snowflake


# Instantiate DAG
dag_instance = customer_analytics_dag()

🛠️ 4. Airflow UI Connection Mapping

To run this pipeline, navigate to the Airflow UI at http://localhost:8080 (Admin -> Connections) and create the following configuration entries:

1. Postgres Hook Connection (my_postgres_conn)

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

2. Amazon Web Services Connection (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"
  }
  

3. Snowflake Connection (my_snowflake_conn)

• Conn Id: my_snowflake_conn
• Conn Type: Snowflake
• Host: account_identifier.snowflakecomputing.com
• Account: account_identifier
• Database: CUSTOMER_DB
• Schema: ANALYTICS
• Login: AIRFLOW_USER
• Password: AIRFLOW_PASSWORD
• Warehouse: COMPUTE_WH
• Role: ANALYTICS_ROLE

4. Slack Connection (slack_conn)

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

❄️ 5. Snowflake Storage Integration & Target Stage DDL

Exposing long-lived AWS Access credentials directly inside Snowflake creates a security vulnerability. The recommended best practice is to configure a Storage Integration allowing Snowflake to authenticate via AWS IAM Roles directly.

Execute the following commands in Snowflake as ACCOUNTADMIN to set up security, tables, and stages:

-- 1. Create a Storage Integration mapping to an IAM Role
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. Retrieve integration properties to trust in AWS
DESCRIBE STORAGE INTEGRATION s3_customer_analytics_int;

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

-- 3. Set up Target Database, Schema, and Tables
CREATE DATABASE IF NOT EXISTS CUSTOMER_DB;
USE DATABASE CUSTOMER_DB;

CREATE SCHEMA IF NOT EXISTS ANALYTICS;
USE SCHEMA ANALYTICS;

-- Target staging table
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 using the Storage Integration credentials
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 schema/role permissions
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;

⚡ 6. Local Quickstart: Launching the Compose Stack

You can run the entire Airflow ecosystem locally using Docker Compose, which configures the Webserver, Scheduler, and PostgreSQL metadata database:

Step 1: Initialize folders and .env

Ensure directory structures exist and setup the Airflow User ID to match the host system permissions:

mkdir -p dags plugins config logs
echo "AIRFLOW_UID=$(id -u)" > .env

Step 2: Boot Docker Compose

docker compose up -d

This automatically runs database migrations (airflow db migrate), constructs the admin account, launches the Airflow Scheduler daemon, and hosts the Airflow Webserver UI locally.

• Airflow Web UI: http://localhost:8080 (Default Login: airflow / airflow)

🚀 7. Deploying to Kubernetes with CeleryExecutor

For high-throughput, horizontally-scalable production environments, aitflow is equipped with configs to deploy to a Kubernetes cluster using the official Apache Airflow Helm chart.

All manifests and helper scripts are situated in the kubernetes/ folder.

Step 1: Run the Deployment Script

Compile the custom application image and trigger the Helm release upgrade:

./kubernetes/deploy.sh

Step 2: Load Image (Minikube Local Development)

If running locally on Minikube, load the newly built Docker image directly into Minikube’s local registry:

minikube image load my-airflow-app:latest

Step 3: Forward Dashboard Ports

To inspect workflows and monitor Celery task queues:

• Airflow Webserver:

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

  Access http://localhost:8080 (Login: admin / admin).

• Flower (Celery Queue Monitor):

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

  Access http://localhost:5555 to view thread metrics, worker states, and active queue processing speeds.

🤝 Next Steps

Check out the repository, submit issues, or fork it to customize the extraction pipelines for your own data platform team: 👉 aitflow Project Template on GitHub