SFTP 到 Google Cloud Storage 传输 Operator¶

Google 提供了一项服务：Google Cloud Storage。该服务用于存储来自各种应用程序的大量数据。SFTP (SSH 文件传输协议) 是一种安全的文件传输协议。它运行在 SSH 协议之上。它支持 SSH 的全部安全和身份验证功能。

先决条件任务¶

要使用这些 Operators，您必须完成以下几项操作：

使用 Cloud Console 选择或创建一个 Cloud Platform 项目。

为您的项目启用结算功能，详情请参阅 Google Cloud 文档。

启用 API，详情请参阅 Cloud Console 文档。
通过 pip 安装 API 库。
pip install 'apache-airflow[google]'
详细信息请参阅安装。
设置 Google Cloud 连接.

Operator¶

SFTP 和 Google Storage 之间的文件传输使用 SFTPToGCSOperator operator 执行。

使用 Jinja 模板配合 source_path, destination_path, destination_bucket, impersonation_chain 动态定义值。

复制单个文件¶

以下 Operator 复制单个文件。

tests/system/google/cloud/gcs/example_sftp_to_gcs.py

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 参数定义了文件在存储桶中的完整路径。

tests/system/google/cloud/gcs/example_sftp_to_gcs.py

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 通配符来复制目录。

tests/system/google/cloud/gcs/example_sftp_to_gcs.py

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 参数中使用 wildcard 通配符来移动特定文件。您在路径中只能使用一个通配符。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 被跳过。

tests/system/google/cloud/gcs/example_sftp_to_gcs.py

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,
)

参考¶

有关更多信息，请参阅

Google Cloud Storage 文档