airflow 运算符从 URL 下载文件并推送到 S3？

Question

是否有气流操作员从 URL 下载 CSV 文件并将文件上传到 S3 ？我可以将本地文件上传到 S3，但想知道是否有操作员可以将文件上传到 S3，而无需将文件下载到我的本地计算机？

Answer 1

我认为最简单的解决方案是使用 on_execute_callback 来执行最简单的任务（在本例中下载文件），并让 S3 操作员上传它。

示例（带 GCS）：

filename = "file-{{ts_nodash}}"
    
def run_and_download_file(context):
    resp = requests.get(".....")
    with open(filename) as f:
        f.write(resp.content)

upload_file = LocalFilesystemToGCSOperator(
    task_id="upload_file",
    src=filename,
    dst=filename,
    bucket=GCS_BUCKET,
    on_execute_callback=run_and_download_backup
)

Answer 2

您可以使用HttpToS3Operator：

http_to_s3_task = HttpToS3Operator(
    task_id="http_to_s3_task",
    http_conn_id=conn_id_name,
    endpoint="/test_file",
    s3_bucket=s3_bucket,
    s3_key=s3_key,
    replace=True,
)

看一下官方文档：
https://airflow.apache.org/docs/apache-airflow-providers-amazon/stable/transfer/http_to_s3.html