Apache Airflow 管理 Amazon Redshift 集群全生命周期Redshift Cluster Operators 与 Sensor 实战指南【免费下载链接】airflowApache Airflow - A platform to programmatically author, schedule, and monitor workflows项目地址: https://gitcode.com/GitHub_Trending/ai/airflowAmazon Redshift 是一项托管的 PB 级数据仓库服务它负责集群的容量供给、监控备份、补丁与引擎升级等全部运维工作让开发者可以专注于数据洞察本身。在 Apache Airflow 中apache-airflow-providers-amazon提供了一组面向 Redshift集群Cluster生命周期管理的 Operators 与 Sensor覆盖创建集群 → 等待可用 → 创建快照 → 暂停 → 恢复 → 删除快照 → 删除集群的完整闭环。本文基于 redshift_cluster.rst 官方指南结合仓库源码、系统测试 DAG 与单元测试逐一对每个 Operator/Sensor 的用法、参数语义与底层实现进行深入讲解读完即可在你的 Airflow DAG 中落地一套可复制的 Redshift 集群自动化管理方案。一、前置准备安装 Provider 与配置 AWS 连接在使用这些 Operators 之前需要完成两件事详见 prerequisite_tasks.rst安装 Amazon Provider它会一并引入 boto3 等 API 库pip install apache-airflow[amazon]配置 AWS 连接Connection参考 AWS Connection 配置指南。默认连接 ID 为aws_default如果机器上存在${HOME}/.aws/下的凭证文件且默认连接的用户名/密码为空Airflow 会自动读取其中的凭证。也可以在连接 Extras 中设置region_name或在环境中导出AWS_DEFAULT_REGION。连接配置细节可继续阅读 aws.rst其中还说明了 MinIO、LocalStack 等 AWS 兼容服务下测试连接结果的正确解读方式。二、通用参数Generic Parameters所有 Redshift 集群相关 Operator/Sensor 都继承自AwsBaseOperator/AwsBaseSensor共享如下通用参数原文见 generic_parameters.rst参数说明默认值aws_conn_id引用 AWS Connection 的 ID。若为None则不查询连接使用 boto3 默认行为分布式部署时需在每台 worker 上维护 boto3 默认配置aws_defaultregion_nameAWS 区域名。为None或省略时使用 AWS Connection Extra 参数中的region_nameNoneverify是否校验 SSL 证书。False表示不校验也可以传入 CA 证书包路径如path/to/cert/bundle.pem。为None时使用连接中的配置Nonebotocore_config用于构造botocore.config.Config的字典可配置重试、超时等例如None{ 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, }注意botocore_config传入空字典{}会覆盖连接级配置省略或传None则使用连接 Extra 中的config_kwargs。三、创建集群RedshiftCreateClusterOperatorRedshiftCreateClusterOperator 用于按指定参数创建一个 Redshift 集群。官方指南中的最小示例摘自系统测试 example_redshift.pycreate_cluster RedshiftCreateClusterOperator( task_idcreate_cluster, cluster_identifierredshift_cluster_identifier, vpc_security_group_ids[security_group_id], cluster_subnet_group_namecluster_subnet_group_name, publicly_accessibleFalse, cluster_typesingle-node, node_typera3.large, master_usernameDB_LOGIN, master_user_passwordDB_PASS, )3.1 核心参数与默认值对照源码构造函数redshift_cluster.py中RedshiftCreateClusterOperator.__init__可确认如下参数语义参数说明默认值cluster_identifier集群唯一标识必填无node_type集群节点类型如ra3.large、dc2.large等参考 AWS 节点类型文档必填无master_username管理员账号用户名必填无master_user_password管理员账号密码必填无cluster_type集群类型single-node或multi-nodemulti-nodedb_name创建集群时的第一个数据库名devnumber_of_nodes计算节点数multi-node时必填1cluster_security_groups关联的集群安全组列表Nonevpc_security_group_ids关联的 VPC 安全组 ID 列表Nonecluster_subnet_group_name关联的集群子网组名称Noneavailability_zoneEC2 可用区Nonepreferred_maintenance_windowUTC 时间内的自动维护窗口Nonecluster_parameter_group_name关联的参数组名称Noneautomated_snapshot_retention_period自动快照保留天数1manual_snapshot_retention_period手动快照默认保留天数Noneport集群对外端口5439cluster_versionRedshift 引擎版本1.0allow_version_upgrade维护窗口内是否允许大版本升级Truepublicly_accessible是否可从公网访问Trueencrypted是否静态加密Falseenhanced_vpc_routing是否启用增强 VPC 路由Falsekms_key_id加密密钥 KMS Key IDNoneiam_roles集群可用来访问其他 AWS 服务的 IAM 角色列表Nonetags标签列表Nonewait_for_completion是否等待集群进入available状态Falsemax_attempt轮询最大尝试次数5poll_interval两次轮询间隔秒数60deferrable是否以 deferrable 模式运行取配置operators.default_deferrabledelete_cluster_on_failure创建后校验失败时是否尽力删除集群Truecleanup_timeout_seconds失败清理删除的超时秒数3003.2 底层调用链与失败清理机制从源码可以看到execute()会把这些 Python 参数映射为 boto3create_clusterAPI 的请求参数如DBName、ClusterType、NumberOfNodes、VpcSecurityGroupIds、Port等。其中一处值得注意的实现细节是PubliclyAccessible始终会被显式写入请求源码注释说明Redshift 侧默认即为 True因此无论取值如何都要显式设置避免依赖 API 默认值。创建完成后若deferrableTrue任务会通过self.defer(...)交由 RedshiftCreateClusterTrigger 轮询集群状态Worker 槽位得以释放若wait_for_completionTrue非 deferrable则同步调用 boto3 waitercluster_available以poll_interval为 Delay、max_attempt为 MaxAttempts 等待集群可用失败清理如果集群已成功发起创建、但后续等待阶段抛WaiterError如 IAM/权限问题且delete_cluster_on_failureTrue会触发有界的最佳努力删除_attempt_cleanup_with_retry60 秒重试间隔、总时长受cleanup_timeout_seconds限制遇到InvalidClusterStateFault/InvalidClusterState会持续重试同时不会掩盖原始异常。单元测试 test_redshift_cluster.py 验证了单节点集群创建时实际传给create_cluster的参数集合DBNamedev、ClusterTypesingle-node、AutomatedSnapshotRetentionPeriod1、ClusterVersion1.0、Port5439等多节点场景则会额外携带NumberOfNodes。四、等待集群状态RedshiftClusterSensorRedshiftClusterSensor 用于轮询集群状态直到达到目标状态。官方示例wait_cluster_available RedshiftClusterSensor( task_idwait_cluster_available, cluster_identifierredshift_cluster_identifier, target_statusavailable, poke_interval15, timeout60 * 30, )cluster_identifier被探测的集群标识target_status期望达到的目标状态默认available系统测试中还用到了paused见 example_redshift.py 中的wait_cluster_pausedpoke_interval/timeout轮询间隔与总超时继承自AwsBaseSensor的通用行为。4.1 实现原理poke()每次调用RedshiftHook.cluster_status()见 hooks/redshift_cluster.py底层即 boto3describe_clusters返回的ClusterStatus若集群不存在则返回特殊值cluster_not_found。命中目标状态即返回True。该 Sensor 同样支持 deferrable 模式deferrableTrue时先执行一次poke未达目标则defer给 RedshiftClusterTrigger 在 Triggerer 侧异步轮询Worker 槽位不会被长时间占用。五、创建与删除集群快照5.1 RedshiftCreateClusterSnapshotOperatorRedshiftCreateClusterSnapshotOperator 为指定集群创建手动快照。官方示例create_cluster_snapshot RedshiftCreateClusterSnapshotOperator( task_idcreate_cluster_snapshot, cluster_identifierredshift_cluster_identifier, snapshot_identifierredshift_cluster_snapshot_identifier, poll_interval30, max_attempt100, retention_period1, wait_for_completionTrue, )关键参数snapshot_identifier快照唯一标识必填retention_period手动快照保留天数-1表示永久保留默认值示例中设为1天wait_for_completion是否等待快照进入available状态默认Falsepoll_interval/max_attempt轮询间隔与最大尝试次数默认15秒 /20次deferrable是否 deferrable 运行。实现要点源码可见前置校验execute()首先通过hook.cluster_status()确认集群处于available状态否则直接抛出AirflowException——快照只能对可用集群创建底层调用RedshiftHook.create_cluster_snapshot()hooks/redshift_cluster.py映射为 boto3create_cluster_snapshotManualSnapshotRetentionPeriod、Tagsdeferrable 模式下委托 RedshiftCreateClusterSnapshotTrigger 轮询且defer时显式设置timeoutmax_attempt * poll_interval 60防止 Trigger 死亡后超时被重置。5.2 RedshiftDeleteClusterSnapshotOperatorRedshiftDeleteClusterSnapshotOperator 删除指定的手动快照。官方示例delete_cluster_snapshot RedshiftDeleteClusterSnapshotOperator( task_iddelete_cluster_snapshot, cluster_identifierredshift_cluster_identifier, snapshot_identifierredshift_cluster_snapshot_identifier, )wait_for_completion默认True删除后通过get_status()底层describe_cluster_snapshots快照不存在返回None持续轮询直至快照消失轮询间隔由poll_interval控制默认10秒。六、暂停与恢复集群为节省成本Redshift 允许暂停/恢复集群。Airflow 提供了对应的两个 Operator并且都支持 deferrable 模式——文档明确指出设置deferrableTrue后任务会从 Worker 槽位中释放状态轮询转移到 Trigger 上进行。6.1 RedshiftPauseClusterOperatorRedshiftPauseClusterOperator 用于暂停处于available状态的集群。官方示例pause_cluster RedshiftPauseClusterOperator( task_idpause_cluster, cluster_identifierredshift_cluster_identifier, )参数wait_for_completion默认False为True时等待集群进入paused、poll_interval默认30秒、max_attempts默认30次、deferrable。源码中的关键防御逻辑构造函数里维护了_remaining_attempts 10与_attempt_interval 15两个内部参数用来应对 boto3 API 的一个已知问题——API 会过早地报告集群可接收请求导致最初的 pause 调用被以InvalidClusterStateFault拒绝。execute()会循环重试最多 10 次、每次间隔 15 秒全部失败才抛出异常。等待完成时deferrable 模式先快速检查一次状态若集群正在deleting则直接报错此时无法暂停否则defer给 RedshiftPauseClusterTrigger非 deferrable 模式则使用 boto3 waitercluster_paused。6.2 RedshiftResumeClusterOperatorRedshiftResumeClusterOperator 用于恢复被暂停的集群。官方示例resume_cluster RedshiftResumeClusterOperator( task_idresume_cluster, cluster_identifierredshift_cluster_identifier, )与 Pause 相同它也内置了 10 次 × 15 秒的InvalidClusterStateFault重试防御wait_for_completionTrue时deferrable 模式检查到available即成功返回、检查到deleting即抛错、否则defer给 RedshiftResumeClusterTrigger非 deferrable 模式使用 waitercluster_resumed。七、删除集群RedshiftDeleteClusterOperatorRedshiftDeleteClusterOperator 删除指定集群同样支持 deferrable 模式。官方示例delete_cluster RedshiftDeleteClusterOperator( task_iddelete_cluster, cluster_identifierredshift_cluster_identifier, )参数说明skip_final_cluster_snapshot删除时是否跳过创建最终快照默认True即不保留最终快照直接删除final_cluster_snapshot_identifier若保留最终快照则指定其名称wait_for_completion默认True等待删除完成poll_interval默认30秒max_attempts默认30次——按源码注释与poll_interval组合后默认给出约15 分钟的等待窗口足以熬过 pause/resize 等中间状态。7.1 忙碌状态重试与 deferrable 双阶段删除非 deferrable 模式execute()循环调用hook.delete_cluster()若抛出InvalidClusterStateFault集群正处于状态转换中会读取当前ClusterStatus并记录日志间隔poll_interval重试最多max_attempts次成功后若wait_for_completionTrue则使用 waitercluster_deleted等待删除完成。deferrable 模式_delete_or_defer_until_settled先尝试发起一次删除若被InvalidClusterStateFault拒绝则defer给 RedshiftClusterSettledTrigger待集群进入可删除状态后由回调_retry_delete_when_settled再次发起删除若删除已被接受则先快速检查集群是否已不存在cluster_not_found即成功返回否则defer给 RedshiftDeleteClusterTrigger 等待删除完成。整个流程的timeout同样为max_attempts * poll_interval 60。八、组装一个完整的集群生命周期 DAG官方系统测试 example_redshift.py 演示了完整的最佳实践编排把上述组件串成一个example_redshiftDAGscheduleonce、start_datedatetime(2021, 1, 1)、catchupFalse。其依赖链条chain与TriggerRule.ALL_DONE组合为create_cluster → wait_cluster_available → create_cluster_snapshot → wait_cluster_available_before_pause → pause_cluster → wait_cluster_paused → resume_cluster → wait_cluster_available_after_resume → 执行 RedshiftDataOperator 写入数据 → delete_cluster_snapshot → delete_cluster值得借鉴的三个编排要点每个集群状态变更后都紧跟一个RedshiftClusterSensor创建后等待available、暂停后等待paused、恢复后再等待available保证下游操作如快照、数据写入不会在错误的状态下执行删除任务使用TriggerRule.ALL_DONEdelete_cluster.trigger_rule TriggerRule.ALL_DONE无论主链路成功或失败都执行清理测试 DAG 中还将delete_cluster.max_attempts调高到 50并在末尾挂watcher任务来正确标记测试的成败删除顺序上先删快照再删集群delete_cluster_snapshot在delete_cluster之前避免遗留手动快照导致额外的存储费用。另外该 DAG 中还演示了用 RedshiftDataOperator 在集群创建完成后执行建表与数据写入将集群生命周期管理与集群内数据操作打通——读者可将其中fruit建表/插入部分替换为自己的业务 SQL。九、deferrable 模式小结本指南涉及的大部分 Operator/Sensor 都支持deferrableTrue其价值在于任务被defer后立即释放 Worker 槽位轮询工作由 Triggerer 中的 [Trigg【免费下载链接】airflowApache Airflow - A platform to programmatically author, schedule, and monitor workflows项目地址: https://gitcode.com/GitHub_Trending/ai/airflow创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考