Amazon Redshift（集群）¶

Amazon Redshift 管理数据仓库的所有设置、操作和扩展工作：配置容量、监控和备份集群，以及将补丁和升级应用于 Amazon Redshift 引擎。您可以专注于使用数据为您的业务和客户获取新的见解。

先决条件任务¶

要使用这些操作符，您必须完成以下几项操作

使用 AWS 控制台或 AWS CLI 创建必要的资源。
通过 pip 安装 API 库。
pip install 'apache-airflow[amazon]'
详细信息请参考 Airflow® 的安装
设置连接.

操作符¶

创建 Amazon Redshift 集群¶

要使用指定的参数创建 Amazon Redshift 集群，可以使用 RedshiftCreateClusterOperator。

tests/system/amazon/aws/example_redshift.py

create_cluster = RedshiftCreateClusterOperator(
    task_id="create_cluster",
    cluster_identifier=redshift_cluster_identifier,
    vpc_security_group_ids=[security_group_id],
    cluster_subnet_group_name=cluster_subnet_group_name,
    publicly_accessible=False,
    cluster_type="single-node",
    node_type="dc2.large",
    master_username=DB_LOGIN,
    master_user_password=DB_PASS,
)

恢复 Amazon Redshift 集群¶

要恢复“已暂停”的 Amazon Redshift 集群，可以使用 RedshiftResumeClusterOperator。您还可以通过将 deferrable 参数设置为 True，以可延迟模式运行此操作符。这将确保任务从 Airflow 工作进程槽中延迟，并且在触发器上发生任务状态的轮询。

tests/system/amazon/aws/example_redshift.py

resume_cluster = RedshiftResumeClusterOperator(
    task_id="resume_cluster",
    cluster_identifier=redshift_cluster_identifier,
)

暂停 Amazon Redshift 集群¶

要暂停 available 的 Amazon Redshift 集群，可以使用 RedshiftPauseClusterOperator。您还可以通过将 deferrable 参数设置为 True，以可延迟模式运行此操作符。

tests/system/amazon/aws/example_redshift.py

pause_cluster = RedshiftPauseClusterOperator(
    task_id="pause_cluster",
    cluster_identifier=redshift_cluster_identifier,
)

创建 Amazon Redshift 集群快照¶

要创建 Amazon Redshift 集群快照，可以使用 RedshiftCreateClusterSnapshotOperator

tests/system/amazon/aws/example_redshift.py

create_cluster_snapshot = RedshiftCreateClusterSnapshotOperator(
    task_id="create_cluster_snapshot",
    cluster_identifier=redshift_cluster_identifier,
    snapshot_identifier=redshift_cluster_snapshot_identifier,
    poll_interval=30,
    max_attempt=100,
    retention_period=1,
    wait_for_completion=True,
)

删除 Amazon Redshift 集群快照¶

要删除 Amazon Redshift 集群快照，可以使用 RedshiftDeleteClusterSnapshotOperator

tests/system/amazon/aws/example_redshift.py

delete_cluster_snapshot = RedshiftDeleteClusterSnapshotOperator(
    task_id="delete_cluster_snapshot",
    cluster_identifier=redshift_cluster_identifier,
    snapshot_identifier=redshift_cluster_snapshot_identifier,
)

删除 Amazon Redshift 集群¶

要删除 Amazon Redshift 集群，可以使用 RedshiftDeleteClusterOperator。您还可以通过将 deferrable 参数设置为 True，以可延迟模式运行此操作符。

tests/system/amazon/aws/example_redshift.py

delete_cluster = RedshiftDeleteClusterOperator(
    task_id="delete_cluster",
    cluster_identifier=redshift_cluster_identifier,
)

传感器¶

等待 Amazon Redshift 集群状态¶

要检查 Amazon Redshift 集群的状态，直到它达到目标状态或另一个终端状态，可以使用 RedshiftClusterSensor。

tests/system/amazon/aws/example_redshift.py

wait_cluster_available = RedshiftClusterSensor(
    task_id="wait_cluster_available",
    cluster_identifier=redshift_cluster_identifier,
    target_status="available",
    poke_interval=15,
    timeout=60 * 30,
)

参考¶

Amazon Redshift 的 AWS boto3 库文档