我目前正在尝试使用 Airflow 中的 FTPHook 来将文件上传到远程 ftp 或从远程 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
感谢您的帮助!
最佳答案
我找到了一个简单的流媒体解决方案,可以使用 pysftp 从 gcs 上传/下载到 ftp 服务器,反之亦然,我想与您分享。 首先,我找到了this解决方案,效果很好,但该解决方案的唯一问题是它不支持从 gcs 上传文件到 FTP。所以我在寻找别的东西。 所以我正在研究不同的方法,所以我找到了这个谷歌 document这基本上允许您流式传输到/从 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)
关于google-cloud-storage - 如何使用 Airflow FTPHook 从 GCS 上传/下载文件到/从 ftp 服务器,我们在Stack Overflow上找到一个类似的问题: https://stackoverflow.com/questions/51642327/