如何解决Airflow 使用 dag python 文件扮演 aws 角色
我正在开发一个 Airflow 管道,一旦传感器任务在特定存储桶路径中找到文件 _SUCCESS
,就会触发 Spark Databricks 作业。
问题是 Airflow 无法直接访问存储桶以验证此文件是否存在,而且我们不是此 AWS 帐户的所有者,但我们拥有允许我们访问此存储桶的角色。
我尝试将来自 S3KeySensor Operator 的 aws_conn_id 与 Airflow 连接一起使用,但它不起作用。 下图显示了 Airflow 中的配置:
with DAG(...) as ingestion_xxx:
staging = DatabricksRunNowOperator(
task_id="xxx-table-staging",job_id=xxx,json={"jar_params": [...]}
)
sensor = S3KeySensor(
task_id="sensor",bucket_name="mybucket",bucket_key="bucketprefix/date={{ ds }}/_SUCCESS",aws_conn_id="aws_s3_role_conn"
)
sensor >> staging
我没有出错,但我得出的结论是问题出在角色上。
注意:我测试了 databricks 作业,它可以正常工作。
如果我们不能使用 Airflow 连接,我们可以使用带有 boto3 的 AWS API 直接从我的 DAG 文件中承担角色吗?
版权声明:本文内容由互联网用户自发贡献,该文观点与技术仅代表作者本人。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如发现本站有涉嫌侵权/违法违规的内容, 请发送邮件至 dio@foxmail.com 举报,一经查实,本站将立刻删除。