1. 项目概述当 Airflow 离开本地服务器真正扎根云原生土壤Airflow is on the Cloud——这句标题不是一句口号而是过去三年我亲手把二十多个核心数据管道从自建 Kubernetes 集群迁移到 AWS 托管服务后最真实的体感。它讲的不是“怎么在 EC2 上装个 Airflow”而是如何让 Airflow 彻底卸下运维包袱成为纯粹的 ELT 编排引擎。关键词里藏着全部真相“Airflow”是调度中枢“ELT Pipeline”是工作负载本质强调 Load 层可计算、可弹性、可版本化“AWS”不是简单部署平台而是整套能力供给方——S3 是数据湖底座Redshift Serverless 或 Athena 是即用即弃的计算层Lambda 和 Step Functions 是轻量胶水逻辑而 Airflow 本身则被剥离成一个专注 DAG 定义、依赖管理、重试策略与可观测性的“纯业务编排层”。这个架构彻底改变了我们团队的工作节奏数据工程师不再花 30% 时间调节点资源、修数据库连接池、救挂掉的 Webserver他们现在每天上午写完一个新 DAG下午就能看到它在生产环境跑通全链路中间不碰任何 infra 配置。适合谁如果你正卡在“Airflow 越用越重、越维护越焦虑”的阶段或者团队刚启动云迁移但还在用 Terraform 硬刚 Airflow Helm Chart又或者你发现每次扩 Redshift 并发数都要改一遍 Airflow 的 connection timeout——那这篇就是为你写的。它不教你怎么写 PythonOperator而是告诉你当 Airflow 真正“on the Cloud”它该长什么样子、不该做什么事、哪些必须交给云服务、哪些必须自己守住。2. 整体架构设计与选型逻辑为什么放弃“云上自建”选择“云原生托管轻量 Airflow”2.1 核心矛盾Airflow 的设计哲学 vs 云环境的天然优势Airflow 最初是为解决“本地批处理任务调度”而生的它假设你有稳定 IP 的 Webserver、可控的元数据库PostgreSQL/MySQL、可预测的 Worker 资源池Celery/K8s。但一上云这些假设全被打破。我亲眼见过三个典型翻车现场场景一用 RDS PostgreSQL 做元数据库DAG 中定义了 50 个 Redshift 连接结果 Airflow Scheduler 每分钟扫一遍所有 connection触发 RDS 连接数暴增最终锁死整个元数据层场景二用 EKS 自建 AirflowWorker Pod 启动要拉镜像初始化 Python 环境加载 DAG平均耗时 42 秒而实际 SQL 任务只跑 8 秒——90% 时间花在“准备干活”上场景三用 Lambda 触发 Airflow DAG通过 REST API但 Lambda 默认超时 15 秒而 Airflow TriggerDagRunEndpoint 在高负载时响应常超 20 秒导致上游事件丢失。这些问题的本质是把“为本地环境优化的调度器”硬塞进“为弹性伸缩设计的云服务”里属于底层范式错配。所以我们的架构第一原则就是Airflow 只做它最擅长的事——DAG 解析、依赖拓扑计算、状态机驱动、失败归因其他一切交给 AWS 原生服务。2.2 关键组件选型每个决策背后都有血泪教训组件我们的选择放弃的方案为什么实测数据支撑Airflow 托管服务Amazon Managed Workflows for Apache Airflow (MWAA)自建 EKS Helm ChartMWAA 控制平面完全托管Scheduler/Webserver/Workers 全免维冷启动时间从 42s→3.2s实测 100 个 DAG 加载RDS 连接池由 MWAA 内置管理connection leak 彻底消失。元数据存储MWAA 内置 Aurora Serverless v2外挂 RDS PostgreSQLAurora Serverless v2 自动扩缩到 0.5 ACU约 1.5GB 内存月均成本 $18而同等 RDS t3.medium 固定实例月均 $32且需手动调优 max_connections、work_mem。对象存储S3分桶前缀隔离EFS/NFS 挂载S3 PUT 操作延迟稳定在 12msp95EFS 在跨 AZ 访问时 p95 延迟达 210msDAG 代码更新后 MWAA 自动同步无需 SSH 登录 Worker 清缓存。计算层Redshift Serverless AthenaEC2 上自建 Presto/TrinoRedshift Serverless 启动计算集群仅需 6 秒vs EC2 预热 3 分钟Athena 按扫描字节数计费单次 10GB 查询成本 $0.05远低于预留 4 核 EC2 的 $0.32/小时。轻量胶水逻辑Lambda Step FunctionsAirflow PythonOperatorLambda 执行 100ms 内函数成本 $0.0000002PythonOperator 在 MWAA Worker 上执行同逻辑平均耗时 850ms且占用 Worker slotStep Functions 状态机可视化依赖图比 Airflow UI 查看嵌套 SubDAG 清晰 3 倍。提示MWAA 不是“把 Airflow 搬上云”而是 AWS 用 3 年重构的云原生调度服务。它禁用BashOperator因无持久化文件系统、强制使用S3DataSyncOperator替代FileTransferOperator、要求 DAG 必须用requirements.txt显式声明依赖——这些“限制”恰恰是云环境的最佳实践入口。2.3 架构全景图数据流与控制流彻底解耦整个 ELT 流程被拆成两条平行线数据流Data PlaneS3 → Redshift/Athena → S3结果表→ BI 工具。所有计算任务直接读写 S3不经过 Airflow Worker。例如COPY FROM S3直接加载原始日志UNLOAD TO S3直接导出聚合结果。控制流Control PlaneMWAA DAG → 触发 Lambda校验 S3 文件清单→ 调用 Redshift Data API执行 SQL→ 轮询 Athena Query State查执行结果→ 更新 Airflow Task State。Airflow 不碰一行数据只发指令、等回调、记日志。这种解耦带来三个质变故障域隔离Redshift 集群升级时Airflow 仍可正常调度其他任务如触发 Lambda 清理临时表不会出现“整个 DAG 被卡住”弹性无瓶颈单个 DAG 可并发触发 50 个 Redshift 查询通过 Data API而传统 Airflow Celery Worker 并发数受限于 RabbitMQ 连接数成本可预测Airflow 成本 MWAA 环境规格 × 运行时长$0.42/小时起计算成本 Redshift/Athena 实际消耗 × 单价$0.064/GB/h 或 $5/TB 扫描没有“闲置 Worker 白白烧钱”。3. 核心细节解析与实操要点从零搭建一个生产级 ELT 编排环境3.1 MWAA 环境创建避开五个致命配置坑MWAA 控制台看似简单但五个参数选错会导致后续所有 DAG 都无法运行DAGs folder必须填s3://your-mwaa-bucket/dags/结尾带/。我曾因少打斜杠MWAA 一直报No module named dags排查 3 小时才发现是 S3 prefix 解析错误。Requirements file路径必须是s3://your-mwaa-bucket/requirements.txt。重点来了不要在这里装boto3、psycopg2、redshift_connectorMWAA 环境已预装这些 AWS SDK 和数据库驱动强行安装会引发版本冲突比如psycopg2-binary2.9.7与 MWAA 内置2.9.5不兼容导致Connection refused。Plugins file留空。MWAA 不支持自定义插件PluginManager 被禁用所有扩展必须用airflow.providers.*包实现。Environment class新手必选mw1.small2 vCPU / 4 GiB RAM。别信文档说“small 够用”实测mw1.small在 50 DAG 场景下 Scheduler CPU 常飙到 95%导致 DAG 解析延迟超 2 分钟。我们线上统一用mw1.medium4 vCPU / 8 GiB RAMScheduler p95 延迟压在 800ms 内。Logging configurationCloudWatch Logs Group 必须手动创建并赋予 MWAA Execution Rolelogs:PutLogEvents权限。否则所有 Task 日志都丢只剩Task instance id failed这句废话。注意MWAA 环境创建后首次启动需 25~40 分钟AWS 在后台部署 Aurora、ECS Tasks、VPC Endpoint。期间控制台显示CREATING但 DAG 上传和触发全部失败——这是正常现象别急着删环境重来。3.2 DAG 设计范式告别“Airflow 万能论”拥抱云服务原生能力我们彻底重构了 DAG 编写逻辑核心口诀是“计算下沉、状态上浮、胶水上云”。以一个典型的用户行为日志 ELT 流程为例原始日志 → 清洗 → 聚合 → 报表# ✅ 正确写法用 AWS 原生 OperatorAirflow 只做 orchestration from airflow import DAG from airflow.providers.amazon.aws.operators.redshift_data import RedshiftDataOperator from airflow.providers.amazon.aws.sensors.s3 import S3KeySensor from airflow.providers.amazon.aws.operators.lambda_function import LambdaInvokeFunctionOperator from datetime import datetime, timedelta default_args { retries: 2, retry_delay: timedelta(minutes5), aws_conn_id: aws_default, # MWAA 预置的 IAM Role 连接 } dag DAG( elg_user_behavior_elt, default_argsdefault_args, schedule_interval0 2 * * *, # 每天凌晨 2 点 start_datedatetime(2024, 1, 1), catchupFalse, tags[elt, redshift], ) # 1. 等待原始日志到达 S3云原生 Sensor wait_for_raw_logs S3KeySensor( task_idwait_for_raw_logs, bucket_keys3://data-lake-raw/logs/{{ ds }}/user_events_*.json, bucket_name, aws_conn_idaws_default, timeout3600, # 1 小时超时避免无限等待 poke_interval300, # 每 5 分钟检查一次 dagdag, ) # 2. 调用 Lambda 校验文件完整性轻量胶水 validate_logs LambdaInvokeFunctionOperator( task_idvalidate_logs, function_namevalidate-s3-objects, payload{ bucket: data-lake-raw, prefix: logs/{{ ds }}/, required_files: [user_events_001.json, user_events_002.json] }, aws_conn_idaws_default, dagdag, ) # 3. 直接执行 Redshift SQL计算下沉 load_to_staging RedshiftDataOperator( task_idload_to_staging, databaseanalytics, sql[ TRUNCATE TABLE staging.user_events;, COPY staging.user_events FROM s3://data-lake-raw/logs/{{ ds }}/ IAM_ROLE arn:aws:iam::123456789012:role/redshift-s3-read FORMAT AS JSON auto; ], aws_conn_idaws_default, dagdag, ) # 4. 聚合计算同样 Redshift 原生 aggregate_metrics RedshiftDataOperator( task_idaggregate_metrics, databaseanalytics, sqlINSERT INTO dws.daily_user_summary SELECT ... FROM staging.user_events WHERE event_date {{ ds }};, aws_conn_idaws_default, dagdag, ) # 依赖链清晰表达数据流向 wait_for_raw_logs validate_logs load_to_staging aggregate_metrics关键设计点解析不用PythonOperator写 SQL过去我们用PythonOperatorredshift_connector执行 SQL结果每次连接 Redshift 都要新建 TCP 连接平均耗时 1.2s而RedshiftDataOperator复用 AWS Data API 的长连接池SQL 提交延迟压到 200ms 内S3KeySensor 替代FileSensorFileSensor依赖本地文件系统MWAA 不提供S3KeySensor直接调用 S3 ListObjectsV2 APIp95 延迟 80msLambda 做校验而非 Airflow校验逻辑如 MD5 校验、JSON Schema 验证用 Lambda 实现执行时间 120ms成本 $0.0000002若用PythonOperatorWorker 启动环境加载执行至少 1.8s还占一个 Worker slot。3.3 安全与权限用最小权限原则锁死每一寸攻击面MWAA 的安全模型极易被低估。我们线上环境严格遵循以下四层隔离网络层MWAA 环境部署在私有子网Private Subnet禁止配置任何公网子网Public Subnet。所有出向流量经 NAT Gateway入向流量仅允许来自 VPC 内部如 Redshift 集群所在安全组。身份层MWAA Execution Role 不直接拥有s3:GetObject权限而是通过AssumeRole 机制委托给具体任务角色。例如RedshiftDataOperator会自动 Assumeredshift-s3-read角色该角色仅允许读取s3://data-lake-raw/*而LambdaInvokeFunctionOperator则 Assumelambda-executor角色仅允许调用指定 Lambda 函数。数据层S3 存储桶启用Bucket Policy ACL 双重控制。Policy 明确拒绝非 VPC 内请求{ Version: 2012-10-17, Statement: [ { Effect: Deny, Principal: *, Action: s3:GetObject, Resource: arn:aws:s3:::data-lake-raw/*, Condition: { StringNotEquals: { aws:SourceVpc: vpc-12345678 } } } ] }审计层CloudTrail 日志捕获所有 MWAA API 调用如CreateEnvironment,UpdateEnvironment并投递到专用 S3 桶开启 S3 服务端加密SSE-KMS和对象版本控制。我们设置 S3 Lifecycle 规则30 天后自动转 Glacier90 天后永久删除确保审计日志不被篡改。实操心得第一次上线时我们忘了给 MWAA Execution Role 添加kms:Decrypt权限导致所有访问 KMS 加密 S3 对象的任务全失败错误日志只显示AccessDeniedException。后来加了一行kms:Decrypt权限问题立刻解决——但花了整整一天排查因为 CloudTrail 日志里没明确提示缺哪个权限。4. 实操过程与核心环节实现从环境创建到第一个 DAG 跑通的完整记录4.1 环境创建全流程含所有 CLI 命令与参数我们不用控制台点点点全部用 AWS CLI CloudFormation 实现 IaCInfrastructure as Code确保环境可复现、可审计。以下是核心步骤第一步创建 S3 存储桶DAG 代码与依赖存放# 创建唯一命名的桶MWAA 要求桶名全球唯一 aws s3 mb s3://mwaa-prod-dags-20240515 --region us-east-1 # 启用版本控制防止误删 DAG aws s3api put-bucket-versioning \ --bucket mwaa-prod-dags-20240515 \ --versioning-configuration StatusEnabled # 设置生命周期规则DAG 代码保留 90 天之后转 Glacier cat lifecycle.json EOF { Rules: [ { Expiration: { Days: 90 }, Status: Enabled, Transitions: [ { Days: 30, StorageClass: GLACIER } ], Prefix: } ] } EOF aws s3api put-bucket-lifecycle-configuration \ --bucket mwaa-prod-dags-20240515 \ --lifecycle-configuration file://lifecycle.json第二步准备 DAG 代码与依赖文件# 创建本地目录结构 mkdir -p mwaa-dags/{dags,plugins,requirements} cd mwaa-dags # 编写 requirements.txt只放 Airflow 特有依赖 echo apache-airflow-providers-amazon8.10.0 requirements.txt echo apache-airflow-providers-redshift4.4.0 requirements.txt # 上传到 S3 aws s3 cp requirements.txt s3://mwaa-prod-dags-20240515/requirements.txt aws s3 cp dags/ s3://mwaa-prod-dags-20240515/dags/ --recursive第三步用 CloudFormation 创建 MWAA 环境关键参数详解# mwaa-env.yaml AWSTemplateFormatVersion: 2010-09-09 Resources: MWAAProductionEnv: Type: AWS::MWAA::Environment Properties: Name: mwaa-prod-20240515 EnvironmentClass: mw1.medium # 强制设为 medium ExecutionRoleArn: !GetAtt MWAARole.Arn DagS3Path: dags/ # 注意结尾斜杠 RequirementsS3Path: requirements.txt SourceBucketArn: !GetAtt MWAADataBucket.Arn NetworkConfiguration: SecurityGroupIds: - !Ref MWAAVPCSecurityGroup SubnetIds: - !Ref PrivateSubnetA - !Ref PrivateSubnetB LoggingConfiguration: DagProcessingLogs: Enabled: true LogLevel: INFO SchedulerLogs: Enabled: true LogLevel: INFO TaskLogs: Enabled: true LogLevel: INFO WebserverLogs: Enabled: true LogLevel: INFO WorkerLogs: Enabled: true LogLevel: INFO提示CloudFormation 模板中DagS3Path必须是相对路径如dags/不能是s3://bucket/dags/。AWS 会自动拼接为s3://bucket/dags/若写全路径会报错Invalid S3 path format。4.2 第一个 DAG 跑通实录从上传到成功执行的每一步验证我们用最简 DAGhello-world-elt.py验证环境连通性from airflow import DAG from airflow.providers.amazon.aws.operators.s3 import S3ListOperator from datetime import datetime dag DAG( hello_world_elt, schedule_intervalNone, start_datedatetime(2024, 1, 1), catchupFalse, ) list_s3 S3ListOperator( task_idlist_s3_objects, bucketmwaa-prod-dags-20240515, prefixdags/, aws_conn_idaws_default, dagdag, )执行步骤与关键观察点上传 DAGaws s3 cp hello-world-elt.py s3://mwaa-prod-dags-20240515/dags/等待同步MWAA 默认每 3 分钟同步一次 S3 DAG可在 CloudWatch Logs 的airflow-scheduler日志组中搜索Syncing DAGs from S3确认触发 DAG在 MWAA 控制台点击Trigger DAG或用 CLIaws mwaa create-cli-token --name mwaa-prod-20240515 # 返回 token 后用 curl 触发MWAA 要求用临时 token 认证 curl -X POST https://mwaa-prod-20240515.region.airflow.amazonaws.com:443/api/v1/dags/hello_world_elt/dagRuns \ -H Authorization: Bearer token \ -H Content-Type: application/json \ -d {conf:{}}验证执行进入 Airflow UI →Graph View看到list_s3_objects节点变绿点击进入Logs应看到类似输出[2024-05-15 08:23:45,123] {s3.py:123} INFO - Listing keys in bucket: mwaa-prod-dags-20240515, prefix: dags/ [2024-05-15 08:23:45,456] {s3.py:127} INFO - Found 3 objects: [dags/hello-world-elt.py, dags/elt_dag.py, dags/__pycache__/]若看到ERROR - Unable to list keys大概率是ExecutionRole缺少s3:ListBucket权限需立即补上。4.3 生产级 ELT DAG 实现以 Redshift Serverless 为例的完整链路真实场景中我们用 Redshift Serverless 替代传统 Redshift 集群DAG 需适配其无服务器特性from airflow import DAG from airflow.providers.amazon.aws.operators.redshift_data import RedshiftDataOperator from airflow.providers.amazon.aws.sensors.redshift_cluster import RedshiftClusterSensor from datetime import datetime, timedelta dag DAG( redshift-serverless-elt, default_args{ retries: 1, retry_delay: timedelta(minutes2), }, schedule_intervalhourly, start_datedatetime(2024, 1, 1), catchupFalse, ) # Redshift Serverless 无需 Sensor无集群启停概念但需确认 Workgroup 存在 # 我们用 Custom Operator 调用 DescribeWorkgroup API 验证 check_workgroup RedshiftDataOperator( task_idcheck_workgroup, databasedev, sqlSELECT 1;, # Serverless Workgroup 无实际 DB此 SQL 仅用于验证连接 workgroup_nameanalytics-workgroup, # Serverless Workgroup 名 aws_conn_idaws_default, dagdag, ) # 加载数据Serverless 使用 Data API无需 JDBC URL load_data RedshiftDataOperator( task_idload_data, databasedev, workgroup_nameanalytics-workgroup, sql[ CREATE TABLE IF NOT EXISTS raw.events (event_id VARCHAR, ts TIMESTAMP);, INSERT INTO raw.events SELECT * FROM s3object s S3 ACCESS_KEY_ID key SECRET_ACCESS_KEY secret SESSION_TOKEN token REGION us-east-1 FORMAT AS JSON auto; ], aws_conn_idaws_default, dagdag, ) check_workgroup load_data关键参数说明workgroup_nameServerless Workgroup 名称替代传统host/portdatabaseServerless 中的数据库名逻辑概念非物理实例S3 ACCESS_KEY_ID必须用临时凭证通过 AssumeRole 获取不可硬编码长期 AKSKREGION必须与 MWAA 环境同 Region跨 Region 访问 S3 会产生额外流量费。5. 常见问题与排查技巧实录那些文档里不会写的血泪经验5.1 DAG 不同步S3 文件已传Airflow UI 却看不到这是新手最高频问题。根本原因只有两个S3 Path 配置错误DagS3Path必须是dags/结尾斜杠若写成dags或dags/但 S3 中实际是s3://bucket/dags/xxx.pyMWAA 会静默失败S3 版本控制未启用MWAA 同步依赖 S3 Object Version ID若桶未开版本控制上传新 DAG 会覆盖旧版本但 MWAA 仍缓存旧版本 ID导致“上传了却看不到”。排查命令# 查看 MWAA 同步日志关键 aws logs filter-log-events \ --log-group-name airflow-mwaa-prod-20240515-scheduler \ --filter-pattern Syncing DAGs # 检查 S3 中 DAG 对象版本 aws s3api list-object-versions \ --bucket mwaa-prod-dags-20240515 \ --prefix dags/ # 强制刷新上传空文件触发同步 aws s3 cp /dev/null s3://mwaa-prod-dags-20240515/dags/.sync-trigger --metadata-directive REPLACE5.2 Task 卡在 “queued” 状态不是资源不足而是权限黑洞当 Task 长时间显示queued超过 5 分钟90% 是权限问题Execution Role 缺少ecs:RunTaskMWAA Worker 本质是 ECS Fargate Task若 Role 无此权限Task 永远无法启动Security Group 出向规则缺失Worker 需访问redshift-data.us-east-1.amazonaws.comData API 端点若 SG 出向只放行0.0.0.0/0但没开 HTTPS443会卡在 DNS 解析VPC Endpoint 配置错误若用 PrivateLink 访问 AWS 服务必须为com.amazonaws.us-east-1.redshift-data创建 VPC Endpoint且 Route Table 要关联。快速验证法# 在 MWAA Worker 上执行需先 SSH 到 Worker EC2但更推荐用 CloudWatch Logs # 查看 Worker 日志中的错误 aws logs filter-log-events \ --log-group-name airflow-mwaa-prod-20240515-worker \ --filter-pattern PermissionDenied|AccessDenied|Failed to resolve5.3 Redshift Data API 调用失败Connection refused 的真实原因错误日志常显示Connection refused但实际与网络无关而是Workgroup 名称大小写敏感Analytics-Workgroup≠analytics-workgroupDatabase 名错误Serverless Workgroup 中database是逻辑名必须与CREATE DATABASE语句一致若从未创建过需先用RedshiftDataOperator执行CREATE DATABASE dev;SQL 语法不兼容Serverless 不支持COPY FROM STDIN必须用COPY FROM S3且指定 IAM Role ARN。终极调试技巧# 用 AWS CLI 直接调用 Data API绕过 Airflow aws redshift-data execute-statement \ --cluster-identifier \ # Serverless 不填此项 --workgroup-name analytics-workgroup \ --database dev \ --sql SELECT 1; \ --region us-east-1若 CLI 成功而 Airflow 失败100% 是 Airflow Operator 参数问题。5.4 成本失控预警那些悄悄烧钱的隐藏项MWAA 环境看似便宜但三个隐藏成本常被忽略S3 LIST 请求费MWAA 每 3 分钟ListObjectsV2一次若 DAG 目录有 1000 个文件每月产生 220 万次 LIST 请求费用约 $22$0.005/1000 次CloudWatch Logs 存储费默认保留 30 天若日志量大如每小时 1GB30 天存储费超 $100Redshift Serverless 休眠费Workgroup 休眠后仍收 $0.001/小时基础费若全天候运行月费 $0.72但若频繁启停API 调用费可能更高。省钱实操用S3 Inventory替代频繁 LIST每天生成一次 CSV 清单Airflow 读 CSV 判断文件是否存在CloudWatch Logs 设置RetentionInDays: 7日志只存 1 周错误日志自动告警正常日志不存Redshift Serverless 开启AutoResume但设置MaxConcurrencyLevel: 1避免多任务并发唤醒。6. 运维与监控让 Airflow on Cloud 真正“无人值守”6.1 关键指标监控只盯这五个 CloudWatch 指标MWAA 将所有指标暴露在 CloudWatch但我们只关注五个决定性指标指标名称命名空间告警阈值说明DagProcessingImportErrorsAWS/MWAA 0DAG 解析失败代码语法错误或 import 缺失SchedulerLatencyAWS/MWAA 10000msScheduler 响应超时通常因 DAG 过多或mw1.small规格不足WorkerUtilizationAWS/MWAA 90%Worker 资源饱和需扩容或优化 DAG 并发RedshiftDataQueryErrorCountAWS/RedshiftData 0Redshift SQL 执行失败需查 SQL 逻辑或权限S3GetObjectErrorCountAWS/S3 0S3 访问失败通常是 IAM Role 权限或 Bucket Policy 问题告警配置 CLI 示例aws cloudwatch put-metric-alarm \ --alarm-name MWAA-Scheduler-Latency-High \ --alarm-description Scheduler latency exceeds 10s \ --metric-name SchedulerLatency \ --namespace AWS/MWAA \ --statistic Average \ --period 300 \ --threshold 10000 \ --comparison-operator GreaterThanThreshold \ --evaluation-periods 2 \ --alarm-actions arn:aws:sns:us-east-1:123456789012:alerts-topic6.2 DAG 版本管理用 Git Commit Hash 锁定生产环境我们禁止直接在 S3 修改 DAG全部走 GitOps每次git push触发 GitHub ActionAction 执行aws s3 sync ./dags/ s3://mwaa-prod-dags-20240515/dags/ --delete同时将git rev-parse HEAD写入s3://mwaa-prod-dags-20240515/version.txtDAG 中添加version_checkTask读取 S3 version.txt 与当前 Git Hash 对比不一致则 fail。这样任何一个生产问题都能精准定位到某次 commit回滚只需git checkout old-commit make deploy。6.3 灾备与恢复当 MWAA 环境崩溃时的 15 分钟重建方案MWAA 环境本身不可备份但我们的灾备策略是DAG 代码 requirements.txt100% 存 GitS3 仅为部署通道MWAA 配置CloudFormation 模板存 Git参数化所有变量如EnvironmentClass,DagS3Path恢复流程aws cloudformation create-stack