【免费下载链接】context-hub项目地址https://gitcode.com/gh_mirrors/co/context-hub点击查看免费下载导读本文基于 apache-airflow-providers-amazon 9.22.0 的维护者文档 编写面向需要在 Apache Airflow DAG 中接入 AWS 服务的开发者。你将掌握该 provider 的约束文件安装方式、aws_default连接的认证配置以及 S3 对象读写、Athena 查询提交、AWS Secrets Manager 密钥解析三大核心工作流的完整可运行示例并了解安装后的验证手段与常见踩坑点。读完即可照着文档在本地或生产环境把 AWS 数据管道跑起来。先决条件三条黄金法则在使用该 provider 之前请先牢记维护者文档强调的三条规则必须与固定的apache-airflow版本一同安装该 provider 不是独立运行时它依赖 Airflow 核心的 hook、operator、sensor 与 secrets backend 框架脱离 Airflow 环境无法单独使用。先建连接、再写 DAG创建一个 Airflow AWS 连接惯例 ID 为aws_default并在写 DAG 代码前让 AWS 凭据可用。凭据绝不写进 DAG 文件账号相关的密钥、IAM 角色假设信息应通过 Airflow 连接Connections、AWS 运行时凭据runtime credentials或密钥后端secret backends提供而不是硬编码在 DAG 源码里。在 Context Hub 仓库 中这篇指南以带 YAML frontmatter 的 Markdown 文档形式存放元数据声明了语言python、版本9.22.0、标签airflow、aws、s3、athena、secrets-manager 等。仓库的 frontmatter 解析器 与 条目注册表 正是读取这类DOC.md文件、将其转换为可检索文档条目的机制——也就是说你可以通过chub get apache-airflow/providers-amazon --lang py之类的方式具体命令以 CLI 参考 为准随时取回这份指南供编码 Agent 使用。安装始终使用官方约束文件官方维护者文档给出的推荐路径是先基于与你的 Airflow 和 Python 版本匹配的官方约束文件constraints file完成 Airflow 安装然后在同一条命令中追加 Amazon provider。这样能避免依赖解析把 Airflow 核心版本“悄悄”升级或降级。python -m venv .venv source .venv/bin/activate python -m pip install --upgrade pip AIRFLOW_VERSIONyour-airflow-version PROVIDER_VERSION9.22.0 PYTHON_VERSION$(python -c import sys; print(f{sys.version_info.major}.{sys.version_info.minor})) CONSTRAINT_URLhttps://raw.githubusercontent.com/apache/airflow/constraints-${AIRFLOW_VERSION}/constraints-${PYTHON_VERSION}.txt python -m pip install \ apache-airflow${AIRFLOW_VERSION} \ apache-airflow-providers-amazon${PROVIDER_VERSION} \ --constraint ${CONSTRAINT_URL}如果 Airflow 已经装好在追加 provider 时必须把 Airflow 包一并钉住防止 pip 自行改动核心版本python -m pip install \ apache-airflowyour-airflow-version \ apache-airflow-providers-amazon9.22.0安装后的两项常规检查airflow providers list | grep amazon airflow info版本提示本文覆盖apache-airflow-providers-amazon版本9.22.0文档元数据versions: 9.22.0、updated-on: 2026-03-12。升级 Airflow 或切换认证模式前请重新核对 provider 安装方式与 AWS 连接文档。该 Provider 常见提供的集成能力维护者文档中明确列出的 AWS 集成主要包括三类分别对应 S3、Athena 与 Secrets Manager类别类名用途S3 文件移动与对象检查S3Hook、S3CreateObjectOperator、S3KeySensor读写 S3 对象、创建对象、等待/校验对象是否存在Athena 查询AthenaOperator在 DAG 中提交 SQL 到 Athena 并将结果写入 S3密钥解析SecretsManagerBackend让 Airflow 从 AWS Secrets Manager 解析连接与变量替代元数据库存储这些类在使用时通过airflow.providers.amazon.aws.*命名空间导入具体路径见下文各工作流的代码示例。认证与配置aws_default连接的两条路线该 provider 的常规 Airflow 连接 ID 是aws_default绝大多数 hook 与 operator 允许通过aws_conn_id参数覆盖默认值。配置方式有两种主流路线。路线一环境变量凭据 Airflow 连接最简单的组合是Airflow AWS 连接只携带区域信息boto3 从运行环境解析凭据。适用于本地开发或 CI 中显式导出密钥的场景export AWS_ACCESS_KEY_IDaccess-key-id export AWS_SECRET_ACCESS_KEYsecret-access-key export AWS_DEFAULT_REGIONus-east-1 export AIRFLOW_CONN_AWS_DEFAULT{conn_type:aws,extra:{region_name:us-east-1}}如果使用临时凭据例如 STS 签发的 session token还需要追加export AWS_SESSION_TOKENsession-token路线二角色托管环境ECS/EKS/EC2/MWAA在 ECS、EKS、EC2、MWAA 或其他基于角色的环境中通常省略静态密钥变量让 boto3 直接使用实例/任务挂载的 IAM 角色。此时连接中只需声明区域凭据由 AWS 运行时自动提供。用 Airflow CLI 创建连接也可以直接用 CLI 把连接写进 Airflowairflow connections add aws_default \ --conn-type aws \ --conn-extra {region_name:us-east-1}当不同 DAG 需要不同账户、区域或 IAM 行为时应使用不同的连接 ID并在任务中显式传aws_conn_id。工作流一读写 S3 对象维护者文档强调应使用 provider 自带的 operators 和 hooks 访问 S3而不是在任务里 shell 调用 AWS CLI。下面是一个完整的 TaskFlow 风格 DAG依次完成“写入 manifest → 等待对象出现 → 读取对象内容”三个步骤from __future__ import annotations import pendulum from airflow import DAG from airflow.decorators import task from airflow.providers.amazon.aws.hooks.s3 import S3Hook from airflow.providers.amazon.aws.operators.s3 import S3CreateObjectOperator from airflow.providers.amazon.aws.sensors.s3 import S3KeySensor with DAG( dag_idaws_s3_demo, start_datependulum.datetime(2024, 1, 1, tzUTC), scheduleNone, catchupFalse, tags[aws, s3], ): write_manifest S3CreateObjectOperator( task_idwrite_manifest, aws_conn_idaws_default, s3_bucketmy-data-bucket, s3_keyincoming/manifest.json, data{run_date: {{ ds }}}, replaceTrue, ) wait_for_manifest S3KeySensor( task_idwait_for_manifest, aws_conn_idaws_default, bucket_namemy-data-bucket, bucket_keyincoming/manifest.json, ) task() def read_manifest() - str: hook S3Hook(aws_conn_idaws_default) body hook.read_key( keyincoming/manifest.json, bucket_namemy-data-bucket, ) print(body) return body write_manifest wait_for_manifest read_manifest()要点说明S3CreateObjectOperator的data参数支持 Jinja 模板如{{ ds }}replaceTrue表示覆盖同名对象S3KeySensor会持续轮询bucket_key直到对象出现常用于等待上游系统落盘在任务代码中S3Hook常用方法包括read_key、load_string、list_keys与check_for_key分别对应读取、写入字符串、列举与存在性检查。工作流二运行 Athena 查询当 DAG 需要向 Athena 提交 SQL 并把结果写回 S3 时使用AthenaOperatorfrom __future__ import annotations import pendulum from airflow import DAG from airflow.providers.amazon.aws.operators.athena import AthenaOperator with DAG( dag_idathena_report_demo, start_datependulum.datetime(2024, 1, 1, tzUTC), scheduleNone, catchupFalse, tags[aws, athena], ): run_query AthenaOperator( task_idrun_query, aws_conn_idaws_default, query SELECT current_date AS run_date, count(*) AS row_count FROM analytics.orders , databaseanalytics, output_locations3://my-athena-results/queries/, workgroupprimary, )运维提醒必须保证运行该任务的 Airflow 身份既拥有启动 Athena 查询的权限也拥有对 Athena 结果 S3 位置的读写权限——这是最容易被遗漏的一环见“常见陷阱”。工作流三从 AWS Secrets Manager 解析连接与变量该 provider 还内置了一个 secrets backend让 Airflow 从 AWS Secrets Manager 读取 Connections 和 Variables而不必只存在 Airflow 元数据库中。在 Airflow 环境中设置后端export AIRFLOW__SECRETS__BACKENDairflow.providers.amazon.aws.secrets.secrets_manager.SecretsManagerBackend export AIRFLOW__SECRETS__BACKEND_KWARGS{connections_prefix:airflow/connections,variables_prefix:airflow/variables,region_name:us-east-1}其中connections_prefix与variables_prefix指定 Secrets Manager 中存放连接与变量的前缀路径region_name指定密钥所在的 AWS 区域。设置完成后通过常规 Airflow API 即可透明地访问from airflow.hooks.base import BaseHook from airflow.models import Variable aws_conn BaseHook.get_connection(aws_default) deploy_env Variable.get(deploy_env) print(aws_conn.conn_id) print(deploy_env)注意修改 secrets backend 配置后需要重启所有应使用新后端的 Airflow 组件调度器、worker、API/Web 进程等否则旧进程仍会沿用此前配置。安装与运行后的验证清单检查项命令provider 已安装airflow providers list | grep amazonAWS 连接存在airflow connections get aws_defaultAirflow 能解析 DAGairflow dags list单任务可导入 provider 类airflow tasks test aws_s3_demo read_manifest 2026-03-12airflow tasks test适合隔离调试单个任务需要调度器、worker 与 secrets backend 端到端参与时应使用正常的 DAG 触发方式。常见陷阱汇总安装时未钉住apache-airflow且未使用官方约束文件可能导致依赖解析把核心版本改动甚至产生不可用的安装。只给一个 Airflow 进程配置 AWS 凭据或 IAM 角色假设调度器、worker、API/Web 进程都需要兼容的 AWS 访问权限。在共享环境里隐式依赖aws_default当 DAG 应指向不同账户或区域时必须显式传aws_conn_id。只给了 Athena 权限却忘了 S3 结果桶权限Athena 查询的结果写入需要额外的 S3 读写权限。在 DAG 文件中硬编码 AWS 密钥或密钥值应改用 Airflow 连接、运行时凭据或 Secrets Manager。结语把这份指南接入你的 Agent 工作流apache-airflow-providers-amazon9.22.0 通过S3Hook/S3CreateObjectOperator/S3KeySensor、AthenaOperator和SecretsManagerBackend覆盖了 AWS 数据管道中最常见的三类需求。本仓库中该指南与 Airflow 核心包指南 同属于 Apache Airflow 文档集前者聚焦 AWS 集成后者覆盖 Airflow 3 安装、DAG 编写与配置基础两者可配合阅读。编写 DAG 前先按本文完成约束文件安装与aws_default连接配置再套用三个工作流模板即可快速落地可运行的 AWS 集成任务。赞分享【免费下载链接】context-hub项目地址https://gitcode.com/gh_mirrors/co/context-hub点击查看免费下载相关推荐ShowDoc 二次开发与贡献指南从环境搭建到 Vue 前端 Slim 4 后端的源码级改造ShowDoc 二次开发与贡献指南从环境搭建到 Vue 前端 Slim 4 后端的源码级改造 本篇指南以 ShowDoc 官方 Development RVC语音转换完整教程10分钟音频训练出你的AI音色模型RVC语音转换完整教程10分钟音频训练出你的AI音色模型 Retrieval based Voice Conversion WebUI下称 RVC是一个开人工智能AI 应用语音音频深度学习Apache Airflow Amazon Provider CLI 参考aws-auth-manager 命令与 Amazon Verified Permissions 实战指南Apache Airflow Amazon Provider CLI 参考aws auth manager 命令与 Amazon Verified Permi后端任务调度工作流自动化数据编排批处理数据工程流程编排上一篇ChatTTS性能优化全攻略vLLM加速、torch.compile与4GB低显存部署技巧下一篇如何快速上手 Bumbag-UIReact 应用开发的完整指南创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考