Amazon Redshift（集群）¶

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

先决条件任务¶

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

使用 AWS Console 或 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 worker 插槽中延迟执行，并在触发器上进行任务状态轮询。

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 库文档