气流操作员从 URL 下载文件并推送到 S3？答案

【问题标题】：airflow operator to download a file from URL and push to S3?气流操作员从 URL 下载文件并推送到 S3？
【发布时间】：2020-05-01 19:48:30
【问题描述】：

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

【问题讨论】：

不，先生。没有这样的运营商。您必须编写一些代码来下载文件。对于上传，您可以利用S3Hook 的load_file() 方法。我不知道有什么方法可以在不下载的情况下将远程文件移动到 S3（当然，除非你可以直接在远程机器上触发 aws s3 cp 命令）
泰。我真正想问的是：是否有一个我可以重复使用的操作员......显然需要将 csv 文件下载到机器上，然后上传到 S3。我只是懒...所以想找一个现成的操作员...

标签： airflow

【解决方案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
)

【讨论】：