SFTP 到 Google Cloud Storage 传输算子¶
Google 有一项服务 Google Cloud Storage。此服务用于存储来自各种应用程序的大量数据。SFTP（SSH 文件传输协议）是一种安全的文件传输协议。它在 SSH 协议上运行。它支持 SSH 的全部安全性和身份验证功能。
前提任务¶
要使用这些算子，您必须执行以下操作
使用 Cloud 控制台 选择或创建 Cloud Platform 项目。
为您的项目启用帐单，如 Google Cloud 文档 中所述。
启用 API，如 Cloud 控制台文档 中所述。
通过 pip 安装 API 库。
pip install 'apache-airflow[google]'详细的信息可以在 安装 中找到。
操作符¶
使用 SFTPToGCSOperator 操作符在 SFTP 和 Google Storage 之间传输文件。
使用 Jinja 模板 和 source_path, destination_path, destination_bucket, impersonation_chain 动态定义值。
复制单个文件¶
以下操作符复制单个文件。
copy_file_from_sftp_to_gcs = SFTPToGCSOperator(
task_id="file-copy-sftp-to-gcs",
source_path=f"{FILE_LOCAL_PATH}/{OBJECT_SRC_1}",
destination_bucket=BUCKET_NAME,
)
移动单个文件¶
要移动文件，请使用 move_object 参数。一旦文件复制到 Google Storage，SFTP 中的原始文件将被删除。 destination_path 参数定义存储桶中文件的完整路径。
move_file_from_sftp_to_gcs_destination = SFTPToGCSOperator(
task_id="file-move-sftp-to-gcs-destination",
source_path=f"{FILE_LOCAL_PATH}/{OBJECT_SRC_2}",
destination_bucket=BUCKET_NAME,
destination_path="destination_dir/destination_filename.bin",
move_object=True,
)
复制目录¶
在 source_path 参数中使用 wildcard 复制目录。
copy_directory_from_sftp_to_gcs = SFTPToGCSOperator(
task_id="dir-copy-sftp-to-gcs",
source_path=f"{FILE_LOCAL_PATH}/{SUBDIR}/*",
destination_bucket=BUCKET_NAME,
)
移动特定文件¶
在 source_path 参数中使用 通配符 来移动特定文件。你只能在路径中使用一个通配符。 destination_path 定义了所有已复制文件的前缀路径，例如 tests_sftp_hook_dir/subdir/parent-1.bin 被复制到 specific_files/parent-1.bin，而 tests_sftp_hook_dir/subdir/parent-2.bin 被复制到 specific_files/parent-2.bin 。 tests_sftp_hook_dir/subdir/parent-3.txt 被跳过。
move_specific_files_from_sftp_to_gcs = SFTPToGCSOperator(
task_id="dir-move-specific-files-sftp-to-gcs",
source_path=f"{FILE_LOCAL_PATH}/{SUBDIR}/*.bin",
destination_bucket=BUCKET_NAME,
destination_path="specific_files/",
move_object=True,
)