Airflow Summit 2026 is coming August 31 - September 2 in Austin, TX. Register now to secure your spot!

Amazon Redshift Data

Amazon Redshift manages all the work of setting up, operating, and scaling a data warehouse: provisioning capacity, monitoring and backing up the cluster, and applying patches and upgrades to the Amazon Redshift engine. You can focus on using your data to acquire new insights for your business and customers.

Prerequisite Tasks

To use these operators, you must do a few things:

Generic Parameters

aws_conn_id

Reference to Amazon Web Services Connection ID. If this parameter is set to None then the default boto3 behaviour is used without a connection lookup. Otherwise use the credentials stored in the Connection. Default: aws_default

region_name

AWS Region Name. If this parameter is set to None or omitted then region_name from AWS Connection Extra Parameter will be used. Otherwise use the specified value instead of the connection value. Default: None

verify

Whether or not to verify SSL certificates.

  • False - Do not validate SSL certificates.

  • path/to/cert/bundle.pem - A filename of the CA cert bundle to use. You can specify this argument if you want to use a different CA cert bundle than the one used by botocore.

If this parameter is set to None or is omitted then verify from AWS Connection Extra Parameter will be used. Otherwise use the specified value instead of the connection value. Default: None

botocore_config

The provided dictionary is used to construct a botocore.config.Config. This configuration can be used to configure Avoid Throttling exceptions, timeouts, etc.

Example, for more detail about parameters please have a look botocore.config.Config
{
    "signature_version": "unsigned",
    "s3": {
        "us_east_1_regional_endpoint": True,
    },
    "retries": {
      "mode": "standard",
      "max_attempts": 10,
    },
    "connect_timeout": 300,
    "read_timeout": 300,
    "tcp_keepalive": True,
}

If this parameter is set to None or omitted then config_kwargs from AWS Connection Extra Parameter will be used. Otherwise use the specified value instead of the connection value. Default: None

Note

Specifying an empty dictionary, {}, will overwrite the connection configuration for botocore.config.Config

Operators

Execute a statement on an Amazon Redshift cluster

Use the RedshiftDataOperator to execute statements against an Amazon Redshift cluster.

This differs from RedshiftSQLOperator in that it allows users to query and retrieve data via the AWS API and avoid the necessity of a Postgres connection.

tests/system/amazon/aws/example_redshift.py[source]

create_table_redshift_data = RedshiftDataOperator(
    task_id="create_table_redshift_data",
    cluster_identifier=redshift_cluster_identifier,
    database=DB_NAME,
    db_user=DB_LOGIN,
    sql=[
        """
        CREATE TABLE IF NOT EXISTS fruit (
        fruit_id INTEGER,
        name VARCHAR NOT NULL,
        color VARCHAR NOT NULL
        );
    """
    ],
    poll_interval=POLL_INTERVAL,
    wait_for_completion=True,
)

Reuse a session when executing multiple statements

Specify the session_keep_alive_seconds parameter on an upstream task. In a downstream task, get the session ID from the XCom and pass it to the session_id parameter. This is useful when you work with temporary tables.

tests/system/amazon/aws/example_redshift.py[source]

create_tmp_table_data_api = RedshiftDataOperator(
    task_id="create_tmp_table_data_api",
    cluster_identifier=redshift_cluster_identifier,
    database=DB_NAME,
    db_user=DB_LOGIN,
    sql=[
        """
        CREATE TEMPORARY TABLE tmp_people (
        id INTEGER,
        first_name VARCHAR(100),
        age INTEGER
        );
    """
    ],
    poll_interval=POLL_INTERVAL,
    wait_for_completion=True,
    session_keep_alive_seconds=600,
)

insert_data_reuse_session = RedshiftDataOperator(
    task_id="insert_data_reuse_session",
    sql=[
        "INSERT INTO tmp_people VALUES ( 1, 'Bob', 30);",
        "INSERT INTO tmp_people VALUES ( 2, 'Alice', 35);",
        "INSERT INTO tmp_people VALUES ( 3, 'Charlie', 40);",
    ],
    poll_interval=POLL_INTERVAL,
    wait_for_completion=True,
    session_id="{{ task_instance.xcom_pull(task_ids='create_tmp_table_data_api', key='session_id') }}",
)

Durable execution

RedshiftDataOperator submits a statement and then polls it to completion on the worker. By default the operator runs in a durable mode that makes this crash-safe: the Redshift statement id is persisted to task state store before polling begins, so if the worker crashes or is preempted and the task is retried, the operator reconnects to the statement that is already executing in Redshift instead of resubmitting the SQL.

On retry the operator checks the prior statement’s state:

  • if it is still running, the operator reconnects and continues polling

  • if it already succeeded, the operator returns immediately without resubmitting

  • if it failed terminally, or its id has expired and is no longer found, the operator submits the SQL fresh

This protection also applies when wait_for_completion=False – even though that task attempt never polls at all, a retry after a successful submission still reconnects rather than resubmitting, since the statement id is persisted immediately after submission regardless of whether the task waits for it to finish.

Durable execution requires Airflow 3.3 or newer, since it relies on the task state store. On earlier Airflow versions the flag is a no-op and the operator always submits fresh SQL on retry, exactly as before. If the task state store is unavailable at runtime, the operator logs that crash recovery is disabled and behaves the same way.

To opt out and always submit fresh SQL on retry, set durable=False:

statement = RedshiftDataOperator(
    task_id="redshift_data",
    database="dev",
    sql="SELECT * FROM table",
    cluster_identifier="cluster_identifier",
    durable=False,
)

Durable execution applies to the synchronous path. When deferrable=True is set, the Triggerer already tracks the statement across the wait, so deferrable mode takes precedence and durable has no effect.

Reference

Was this entry helpful?