如何使用 Airflow FTPHook 从 GCS 向/从 ftp 服务器上传/下载文件答案

【问题标题】：How to upload/download file from GCS to/from ftp server with Airflow FTPHook如何使用 Airflow FTPHook 从 GCS 向/从 ftp 服务器上传/下载文件
【发布时间】：2019-01-09 13:24:36
【问题描述】：

我目前正在尝试在 Airflow 中使用 FTPHook，以便向远程 ftp 上传和下载文件。但我不确定是否可以将 gs:// 路径用作源/目标路径的一部分。我目前不想在 AF pod 中使用本地文件夹，因为文件大小可能会变大，所以我宁愿直接使用 gcs 路径或 gcs 文件流。

conn = FTPHook(ftp_conn_id='ftp_default')
conn.store_file('in', 'gs://bucket_name/file_name.txt')

链接到 FTPHook 代码： here

感谢您的帮助！

【问题讨论】：

标签： google-cloud-storage airflow

【解决方案1】：

我找到了一个简单的流解决方案，可以使用 pysftp 从 gcs 上传/下载到 ftp 服务器，反之亦然，我想与您分享。首先，我找到了this 解决方案，效果很好，但该解决方案的唯一问题是它不支持从 gcs 上传文件到 FTP。所以我在寻找别的东西。因此，与我正在研究不同的方法相比，我发现了这个 google document，它基本上允许您流式传输到 blob 文件/从 blob 文件流式传输，这正是我正在寻找的。p>

        params = BaseHook.get_connection(self.ftp_conn_id)
        cnopts = pysftp.CnOpts()
        cnopts.hostkeys = None
        ftp = pysftp.Connection(host=params.host, username=params.login, password=params.password,
                                 port=params.port,
                                 cnopts=cnopts)

        #This will download file from FTP server to GCS location
        with ftp.open(self.ftp_folder + '/' + file_to_load, 'r+') as remote_file:
            blob = bucket.blob(self.gcs_prefix + file_to_load)
            blob.upload_from_file(remote_file)

        #This will upload file from GCS to FTP server
        with sftp.open(self.ftp_folder + '/' +file_name,'w+') as remote_file:
            blob = bucket.blob(fileObject['name'])
            blob.download_to_file(remote_file)

【讨论】：

【解决方案2】：

GCS 没有实现 FTP 支持，所以这不起作用。看起来 FTP 钩子只知道如何处理本地文件路径或缓冲区，而不是 GCS API 之一。

您也许能够找到（或编写）一些从 FTP 读取并写入 GCS 的代码。

【讨论】：

感谢迈克。我将开始深入研究。
顺便说一句，您可以在不使用临时内存来存储文件的情况下做到这一点吗？由于我的 AF 在 docker 上运行，我不确定如果文件太大，它将如何处理本地存储文件
您必须找到/编写逐块传输的代码，以避免将整个对象加载到内存中。
非常感谢那个迈克。但我确实还有一个问题，因为您正在研究 GCS。为什么在 c# google-cloud-storage 库中获取到 gcs 文件的流非常容易，而 Python 库中不包含它？（我将不得不使用 boto 或实现类似的东西：dev.to/sethmichaellarson/…）