亚马逊云科技 2026 年 9 月 29 日公开一套 MWAA 插件配方：把失败任务的日志与算子脚本送进 Bedrock，单次分析 5 到 10 秒。 这篇配方的三位作者都是亚马逊云科技的交付顾问，正是「前置部署」Forward Deployed Engineer（FDE）的原型岗位：驻在客户现场、把平台能力落成能跑的生产系统，而不是交一份演示。「前置部署」Forward Deployed Engineer（FDE）在数据平台项目里最难写进 SOW 的一段就是 Day 2 运维，也就是管道天天失败、而排查只能靠人翻日志的那一段，这条配方正好补在这里。 Apache Airflow 已经成为数据管道的编排骨干，但当管道长到数百个 DAG、横跨 AWS Glue、Amazon EMR、Amazon Athena 与 Amazon Redshift 时，调试单个任务失败会变成一项很重的运维动作。 亚马逊云科技 2026-09-29T16:17:09Z 在官方大数据博客发布《Building an LLM-powered DAG failure analysis plugin for Amazon MWAA》，给出一条可复现的自建路线。 作者三人是 Sushant Samantaray（资深交付顾问，19 年从业经验）、Parameswara Reddy Gajjela（交付顾问，11 年以上数据分析经验）与 Kamen Sharlandjiev（首席大数据与 ETL 方案架构师，MWAA 与 AWS Glue ETL 方向专家）。 原来的做法是：任务失败后，数据工程师去翻「执行日志」、比对 DAG 配置、逐条分析错误信息，才能找到根因，期间管道服务级别协议被拖慢，团队产能也被占用。 新做法是把这段排查做成 Airflow UI 里的一个按钮：插件自动取上下文、调模型、返回结构化「诊断报告」，单次分析通常 5 到 10 秒完成。 整套流程说清是五步。第一步，使用者在 Airflow UI 里选中一个失败任务，点 Analyze Task，触发一次「失败诊断」。 第二步是上下文收集：插件从 Airflow 的「元数据数据库」取失败「任务实例」的元数据，再从 Amazon S3 取 DAG 源码与日志。 第三步是算子感知增强：按算子类型去取真正失败的那段代码，例如 Glue 作业里的 PySpark 脚本，或 Athena 算子里的 SQL 查询。 第四步把拼好的上下文送进 Amazon Bedrock，第五步把结果渲染回任务实例页面。 从实现上看，这是一条 POST /api/analyze-task 链路：它先从 Airflow 的 aws_default 连接取凭据，组装上下文、调用 fetch_and_add_operator_script 做算子增强、再用 prompts.py 的「提示模板」拼 prompt，最后调 Bedrock 返回结构化分析。 这条路与通用日志分析器最大的差别，是它会按算子类型把「真正失败的代码」一并取来，而不是只把报错信息丢给模型。 外部脚本类算子走 API 回捞：GlueJobOperator 从 AWS Glue 作业定义里找到 PySpark 或 Python 脚本在 Amazon S3 的位置；EmrAddStepsOperator 从 step 参数里取脚本路径；EmrServerlessStartJobOperator 从 job driver 里取 Spark 脚本。 内联类算子直接读算子的渲染模板字段，不产生任何外部 API 调用：AthenaOperator 与 RedshiftDataOperator 取 SQL 查询，BashOperator 取 Bash 命令，PythonOperator 取 DAG 源码里的函数，DBTOperator 同样走内联路线。 路由逻辑落在 script_utils.py 的 process_operator_script 函数里，原文列出的算子共 8 类，覆盖 Glue、EMR、EMR Serverless、Athena、Redshift、Bash、Python 与 DBT。 还有一个省调用的细节：插件实现了智能抓取，只有错误信息里出现 SyntaxError、TypeError、数据类型不匹配这类与代码相关的模式时，才真的去调 S3 取脚本；超时这类基础设施错误会完全跳过脚本抓取。FDE 排障时要记得这条分支：跳过抓取的分析不会给出「代码定位」，那类报告只能当线索用。 这个算子感知的差别直接决定了能不能做「代码定位」：把报错和具体代码行对齐之后，报告才能给出带文件名与行号的根因，而不是复述一遍错误信息。 照搬之前先核对前置条件，缺一项都会卡在部署环节，FDE 可以拿这份清单直接和客户对齐环境要求。 第一，MWAA 环境要跑 Apache Airflow 3.x，走查用的是 Airflow 3.2；插件是通过 Airflow 3.x 才引入的 FastAPI 插件接口 fastapi_apps 注册 UI 的，还在跑 Airflow 2.x 的存量环境不能直接套用。 第二，要让 Amazon Bedrock 在所在区域可用，并开通 Anthropic Claude 模型家族；示例用 Claude，改成 Amazon Nova 只需要改 prompts.py 里的 prompt payload 格式。 第三，MWAA 的执行角色要有 bedrock:InvokeModel 与 s3:GetObject 两个权限。 第四，支撑 MWAA 的 S3 桶必须打开版本控制，因为更新环境时要按对象版本号指定插件包。 第五，本地要有 Python 3.10 或更高版本，以及配置好的 AWS CLI。 第六，多数区域要走推理配置文件 ID，例如 us.anthropic.claude-sonnet-4-5-20250929-v1:0，而不是裸的按需模型 ID；配置前先跑 aws bedrock list-inference-profiles 确认目标模型是 ACTIVE。 部署分五步，全部是可复制的命令，FDE 可以先在开发环境完整走一遍再考虑上生产。 第一步克隆仓库：git clone https://github.com/aws-samples/sample-aws-mwaa-llm-powered-plugin.git 克隆之后不需要改配置：Airflow 会自动发现 plugins 目录下的 AirflowPlugin 子类，不需要任何注册调用；在 MWAA 上，插件文件通过 plugins.zip 投递并解压到 /usr/local/airflow/plugins/。 第二步打包并上传：cd plugins 后 zip -r ../plugins.zip .，再 cd ..，用 aws s3 cp plugins.zip s3://<bucket>/plugins.zip 上传，然后跑 aws s3api head-object --bucket <bucket> --key plugins.zip --query VersionId --output text 记下返回的 VersionId。 这里有一个能省掉整次失败的取舍：插件只依赖 fastapi 与 boto3，而这两个包在 MWAA 的 Airflow 3.x 镜像里已经预装，所以不要写 requirements.txt，少一个依赖文件就少一类包解析冲突导致的「环境更新」失败。 第三步更新环境：aws mwaa update-environment --name <env> --plugins-s3-path plugins.zip --plugins-s3-object-version <上一步的 VersionId>；这两个长参数分别对应 API 里的 plugins_s3_path 与 plugins_s3_object_version 字段，环境会自动重启，通常 10 到 30 分钟。 监控状态用 aws mwaa get-environment --name <env> --query "Environment.{Status:Status,Plugins:PluginsS3Path}" --output json。 第四步配置 Bedrock 连接。MWAA 上 aws_default 连接默认存在且解析到环境执行角色，多数情况不用动；要换区域就编辑 Airflow UI 里 Admin 下的 Connections 中的 aws_default，在 Extra 字段填 {"region_name": "us-east-1"}，login 与 password 留空以继续用执行角色。 第五步验证：等环境更新完成后进 Airflow UI 的 Admin > Plugins，看到 task_analyzer_plugin 就说明「插件注册」成功，任务实例页会出现 Analyze Task 入口。 仓库自带模拟失败的示例 DAG，可以直接验证整条链路。把 dags 目录递归复制到 MWAA 桶的 DAGs 目录，等 1 到 2 分钟让 MWAA 同步。 然后在 UI 里触发 test_aws_sql_operators 让预期的失败发生，打开那个失败的「任务实例」并点 Analyze Task，确认「诊断报告」里同时有带文件与行号的根因、带代码示例的「逐步处置」，以及「预防建议」与监控建议三部分。 插件在把内容发出环境之前做了三层处理，这三层是照搬时必须保留的；「前置部署」Forward Deployed Engineer（FDE）在合规评审里最常被追问的也正是这三层。 第一层是凭据脱敏：sanitize_script 会移除脚本与日志里的密码、令牌、访问密钥等敏感模式。 第二层是内容截断：truncate_script 会把内容压到模型上下文窗口以内。 第三层是路径穿越防护：read_allowlisted_file 先解析规范路径，确认它落在「允许目录」之内才读文件。 如果日志或脚本可能含有个人信息，内置的脱敏不够用：可以在调用 Bedrock 之前加一次 Amazon Comprehend 的 DetectPiiEntities，它会返回实体类型与字符偏移，据此把片段遮蔽掉再送出。代价是每次分析多一次 API 调用与一份成本，所以只在合规要求指向它时才加。 另外，插件端点是靠 Airflow 内置鉴权保护的，没有另建一层授权，这一点在多租户环境里要提前想清楚。 成本主要是 Amazon Bedrock 推理，按每次分析消耗的输入与输出 token 计费：输入来自任务日志、DAG 源码与算子脚本，输出是那份诊断报告。 为控制成本，插件带了一个「缓存机制」：结果以「错误上下文」的哈希为键存下来，同一失败模式重复分析会直接命中缓存，不再调 Bedrock。FDE 在验收时要问清缓存命中的口径，否则账单无法归因到具体团队。 原文在最佳实践里给了四条上线收口动作，FDE 在交付评审里可以逐条核对。 第一条是 IAM 最小权限：只授予所选模型 ID 的 bedrock:InvokeModel，并把 s3:GetObject 收窄到算子脚本真正所在的桶路径。 第二条是数据卫生：配置值放 AWS Secrets Manager，不要硬编码进 DAG 源码。 第三条是运维韧性：在 Bedrock 调用外围加 retry 与熔断，并用 Amazon CloudWatch 监控插件性能、对失败率设告警。 第四条是访问控制：DAG 级别的权限管理可以接原文提到的按标签自动授权方案。 可选的扩展方向有四个：接 Amazon SNS 或 Slack 做主动通知；用 Amazon Bedrock Knowledge Bases 把历史分析做成知识库，让推荐走 RAG；为自己的自定义算子加 handler；对已知失败模式做自动修复，例如调完资源后重启任务。这四类扩展的共同点是，把「智能分析」从单点工具做成平台能力层。 对「前置部署」Forward Deployed Engineer（FDE）来说，这段路由表是整套方案里最值得自己扩展的地方：客户现场真正难处理的是自研算子与厂商私有算子，而 process_operator_script 的路由表正是这些算子的落点。 对「前置部署」Forward Deployed Engineer（FDE）来说，下面这些局限比功能清单更值得先看一遍，因为它们决定这套东西能不能进客户的交付范围。 以下是这份配方没有解决、或原文自己标注为可选的部分，照搬时要把它们一起算进范围。 第一，没有质量数字。原文没有给准确率、召回率或误判率，唯一可量化的效果是「单次分析通常 5 到 10 秒」，而这只是耗时，不是诊断对不对。 第二，没有金额。成本段只给了计费口径，既没给单次分析的 token 用量，也没给按月估算，所以「前置部署」Forward Deployed Engineer（FDE）没法据此直接给客户报价。 第三，5 到 10 秒这个区间没有说明日志规模、上下文长度与所选模型，也没有说明触发截断时耗时怎么变，跨环境外推不可靠。 第四，兼容面很窄：fastapi_apps 是 Airflow 3.x 的插件接口，仍在 Airflow 2.x 的 MWAA 环境要么先升级，要么这套 UI 注册方式整段作废。 第五，只覆盖 8 类算子。自研算子、厂商私有算子，以及 Snowflake、Databricks 这类非 AWS 数据系统都没有 handler，要自己补实现，原文没有给 handler 的编写模板。 第六，安全边界只到「发送前脱敏加截断加路径白名单」。PII 处理是可选项、要额外加 Comprehend 与成本，而且原文没有覆盖数据出境与驻留合规，隔离环境里能不能调 Bedrock 也没有讨论。 第七，这是「样例仓库」而不是产品：没有 SLA、没有版本兼容承诺、没有长期维护说明，作者身份是交付顾问而不是该插件的产品团队。 第八，缓存键是「错误上下文」的哈希，但原文没有给键的粒度、有效期与失效策略，换了模型或改了「提示模板」之后旧结果会不会被复用，需要自己验证。 第九，没有对照组。原文没有给出接入前后的人工排查耗时对比、也没有说明它在多大比例的失败上给出过错误根因；模型给出的是一个看起来完整的「诊断报告」，报告通顺不等于结论正确。 谁该行动、具体怎么做，可以按下面五步走。 第一步，先判断该不该做：跑 MWAA 或自建 Airflow 3.x、DAG 数已经上百、任务失败排查仍靠人翻日志的团队，以及正在客户现场做数据平台 Day 2 运维的「前置部署」Forward Deployed Engineer（FDE），值得先做一个环境试点。 第二步，按部署五步把插件装到开发环境：克隆仓库、打包上传拿 VersionId、用 update-environment 带上 --plugins-s3-object-version 更新环境、按需在 aws_default 里填 {"region_name": "us-east-1"}、最后在 Admin > Plugins 里确认 task_analyzer_plugin 出现。这一步 FDE 不要在客户生产环境做，先要一个独立环境。 第三步，先用仓库自带的 test_aws_sql_operators 跑通一次失败，在任务实例页点 Analyze Task，把「报告里有没有文件名与行号」当作验收标准。如果报告只复述错误信息、给不出「代码定位」，先查 process_operator_script 的算子路由，大概率是这类算子没有 handler 或脚本没被取到，而不是模型不行。 第四步，上线前做三件收口的事：执行角色只给 bedrock:InvokeModel 与限定路径的 s3:GetObject；在 Bedrock 调用外围加 retry 与熔断，并在 CloudWatch 上对失败率设告警；若日志可能含个人信息，在调用前加一次 Comprehend 的 DetectPiiEntities。 第五步，把变更留成可回滚的形态：记录每次发布的插件包对象版本，回滚就是把 --plugins-s3-object-version 指回上一个 VersionId 再更新环境，DAG 完全不用改；彻底退出时先 aws s3 rm 删掉桶里的 plugins.zip，再更新环境去掉插件引用并等它重启，最后视情况从执行角色里移除 Bedrock 权限。 适用条件只有三条：Airflow 升到 3.x；Bedrock 在所在区域可用且已开通 Claude 或 Nova；MWAA 执行角色允许挂上 bedrock:InvokeModel 与 s3:GetObject。三条里缺任何一条，都不要把这份配方写进客户的交付计划。 最后一条判断规则：这套方案省下的是「翻日志」的时间，不是判断责任。「前置部署」Forward Deployed Engineer（FDE）该做的是把「报告能不能指到代码行」定为验收条件、把「诊断是否正确」留成人来签字的环节，而不是把模型的输出直接接到自动修复上。 原始来源： - https://aws.amazon.com/blogs/big-data/building-an-llm-powered-dag-failure-analysis-plugin-for-amazon-mwaa/ - https://github.com/aws-samples/sample-aws-mwaa-llm-powered-plugin