Data Engineering Path · Airflow
AWS CORE PLATFORM CASE STUDY
Redshift Operators & Hooks
The Warehouse at the End of the Pipeline
Redshift shows up as the final destination in a huge share of real Airflow pipelines — the case study elsewhere in this course loads exactly this way. This page is the dedicated reference for everything Redshift-specific: running SQL, pausing/resuming clusters to control cost, and waiting on cluster state.
A Note on This Page's Screenshots
Every other AWS page in this section runs its code against real, live AWS resources created specifically for this course. A Redshift cluster is different: even the smallest provisioned cluster costs real money to run and takes several minutes to spin up, and this AWS account has no existing cluster or Redshift Serverless workgroup to reuse. The code below is correct, real, production-representative Redshift provider code — it's simply not been executed live for a screenshot, unlike the S3/SNS/SQS/Lambda/DynamoDB/Athena pages.
Every other AWS page in this section runs its code against real, live AWS resources created specifically for this course. A Redshift cluster is different: even the smallest provisioned cluster costs real money to run and takes several minutes to spin up, and this AWS account has no existing cluster or Redshift Serverless workgroup to reuse. The code below is correct, real, production-representative Redshift provider code — it's simply not been executed live for a screenshot, unlike the S3/SNS/SQS/Lambda/DynamoDB/Athena pages.
Running SQL Against Redshift
Two ways to run SQL, depending on whether you want a persistent JDBC-style connection or the newer Data API (no VPC networking required — calls go through the AWS API, not a direct DB connection):
from airflow.providers.amazon.aws.operators.redshift_data import RedshiftDataOperator
run_transform = RedshiftDataOperator(
task_id="run_daily_transform",
cluster_identifier="analytics-cluster",
database="analytics",
sql="""
INSERT INTO fact_daily_sales
SELECT order_date, SUM(amount) AS total_amount
FROM staging_orders
WHERE order_date = '{{ ds }}'
GROUP BY order_date;
""",
aws_conn_id="aws_default",
)
from airflow.providers.common.sql.operators.sql import SQLExecuteQueryOperator
run_transform_jdbc = SQLExecuteQueryOperator(
task_id="run_daily_transform_jdbc",
conn_id="redshift_default", # a Connection of type Amazon Redshift
sql="INSERT INTO fact_daily_sales SELECT ... ;",
)
RedshiftDataOperator vs SQLExecuteQueryOperator
RedshiftDataOperator uses the Redshift Data API — no persistent connection, no VPC access needed from the Airflow worker, just IAM permissions. SQLExecuteQueryOperator needs a real JDBC connection, meaning the worker must have network access into the cluster's VPC. The Data API route is usually simpler for Airflow running outside the cluster's own network.
Controlling Cluster Cost: Pause & Resume
Provisioned Redshift clusters bill by the hour whether they're doing anything or not — a common cost-control pattern is resuming the cluster right before a nightly load and pausing it again once done:
from airflow.providers.amazon.aws.operators.redshift_cluster import (
RedshiftResumeClusterOperator,
RedshiftPauseClusterOperator,
)
from airflow.providers.amazon.aws.sensors.redshift_cluster import RedshiftClusterSensor
resume_cluster = RedshiftResumeClusterOperator(
task_id="resume_cluster",
cluster_identifier="analytics-cluster",
aws_conn_id="aws_default",
)
wait_for_available = RedshiftClusterSensor(
task_id="wait_for_cluster_available",
cluster_identifier="analytics-cluster",
target_status="available",
aws_conn_id="aws_default",
poke_interval=30,
)
# ... run transform goes here ...
pause_cluster = RedshiftPauseClusterOperator(
task_id="pause_cluster",
cluster_identifier="analytics-cluster",
aws_conn_id="aws_default",
)
resume_cluster >> wait_for_available >> run_transform >> pause_cluster
RedshiftSQLHook Directly
from airflow.providers.amazon.aws.hooks.redshift_sql import RedshiftSQLHook
def check_row_count(**context):
hook = RedshiftSQLHook(redshift_conn_id="redshift_default")
count = hook.get_first("SELECT COUNT(*) FROM fact_daily_sales WHERE order_date = %s", parameters=[context["ds"]])[0]
if count == 0:
raise ValueError("No rows loaded for today - upstream extract likely failed")
Redshift Serverless Changes This Calculus
Redshift Serverless (billed per RPU-second of actual query activity, no cluster to pause/resume) removes the need for the resume/pause pattern above entirely — if cost is the main driver behind pause/resume, evaluate whether Serverless fits before building that orchestration complexity into the DAG.
Redshift Serverless (billed per RPU-second of actual query activity, no cluster to pause/resume) removes the need for the resume/pause pattern above entirely — if cost is the main driver behind pause/resume, evaluate whether Serverless fits before building that orchestration complexity into the DAG.
arrow_back Previous Topic
Athena Operators and Hooks
Next Topic arrow_forward
EMR Operators and Hooks