我一直在尝试使用 Airflow 来安排 DAG。 DAG 之一包括一个从 s3 存储桶加载数据的任务。出于上述目的，我需要设置 s3 连接。但是气流提供的用户界面并不那么直观（ http://pythonhosted.org/airflow/configuration.html?highlight=connection#connections ）。任何人都成功设置了 s3 连接，如果有的话，你们有什么最佳实践吗？谢谢。原文由 Nikhil Reddy 发布，翻译遵循 CC BY-SA 4.0 许可协议

假设气流托管在 EC2 服务器上。只需按照其他答案创建连接，但除了应保留为 S3 的连接类型外，将配置中的所有内容留空 S3hook 将默认为 boto，这将默认为您运行 airflow 的 EC2 服务器的角色。假设此角色有权访问 S3，您的任务将能够访问存储桶。这是比使用和存储凭据更安全的方法。原文由 ninjasmith 发布，翻译遵循 CC BY-SA 3.0 许可协议

使用 UI 的 Airflow s3 连接

编辑：此答案以明文形式存储您的密钥，这可能存在 安全风险，不建议这样做。最好的方法是将访问密钥和密钥放在登录/密码字段中，如下面其他答案所述。结束编辑

很难找到参考资料，但经过一番挖掘后，我能够让它发挥作用。

TLDR

创建具有以下属性的新连接：

连接 ID： my_conn_S3

连接类型： S3

额外的：

 {"aws_access_key_id":"_your_aws_access_key_id_", "aws_secret_access_key": "_your_aws_secret_access_key_"}

长版本，设置 UI 连接：

在 Airflow UI 上，转到 Admin > Connections
创建具有以下属性的新连接：
连接 ID： my_conn_S3
连接类型： S3
额外： {"aws_access_key_id":"_your_aws_access_key_id_", "aws_secret_access_key": "_your_aws_secret_access_key_"}
将所有其他字段（主机、模式、登录）留空。

要使用此连接，您可以在下面找到一个简单的 S3 传感器测试。这个测试的想法是设置一个传感器来监视 S3 中的文件（T1 任务），一旦满足以下条件，它就会触发 bash 命令（T2 任务）。

测试

在运行 DAG 之前，请确保您有一个名为“S3-Bucket-To-Watch”的 S3 存储桶。
添加下面的 s3_dag_test.py 到 airflow dags 文件夹 (~/airflow/dags)
开始 airflow webserver 。
转到 Airflow 用户界面 (http://localhost:8383/)
开始 airflow scheduler 。
在主 DAG 视图上打开“s3_dag_test”DAG。
选择“s3_dag_test”以显示 dag 详细信息。
在图表视图上，您应该能够看到它的当前状态。
‘check_s3_for_file_in_s3’ 任务应该处于活动状态并正在运行。
现在，将名为“file-to-watch-1”的文件添加到“S3-Bucket-To-Watch”。
第一个任务应该已经完成，第二个应该开始并完成。

dag 定义中的 schedule_interval 设置为 ‘@once’，以方便调试。

要再次运行它，请保留所有内容，删除存储桶中的文件，然后通过选择第一个任务（在图表视图中）并选择“清除”所有“过去”、“未来”、“上游”、“下游”来重试…. 活动。这应该会再次启动 DAG。

让我知道进展如何。

s3_dag_test.py ;

 """
S3 Sensor Connection Test
"""

from airflow import DAG
from airflow.operators import SimpleHttpOperator, HttpSensor,   BashOperator, EmailOperator, S3KeySensor
from datetime import datetime, timedelta

default_args = {
    'owner': 'airflow',
    'depends_on_past': False,
    'start_date': datetime(2016, 11, 1),
    'email': ['something@here.com'],
    'email_on_failure': False,
    'email_on_retry': False,
    'retries': 5,
    'retry_delay': timedelta(minutes=5)
}

dag = DAG('s3_dag_test', default_args=default_args, schedule_interval= '@once')

t1 = BashOperator(
    task_id='bash_test',
    bash_command='echo "hello, it should work" > s3_conn_test.txt',
    dag=dag)

sensor = S3KeySensor(
    task_id='check_s3_for_file_in_s3',
    bucket_key='file-to-watch-*',
    wildcard_match=True,
    bucket_name='S3-Bucket-To-Watch',
    s3_conn_id='my_conn_S3',
    timeout=18*60*60,
    poke_interval=120,
    dag=dag)

t1.set_upstream(sensor)

主要参考资料：

原文由 Anselmo 发布，翻译遵循 CC BY-SA 4.0 许可协议

使用 UI 的 Airflow s3 连接

TLDR

长版本，设置 UI 连接：

测试

s3_dag_test.py ;

你尚未登录，登录后可以

Stack Overflow 翻译