SFTP 到 Google Cloud Storage 传输操作符¶

Google 有一项服务 Google Cloud Storage。此服务用于存储来自各种应用程序的大量数据。SFTP（SSH 文件传输协议）是一种安全的文件传输协议。它通过 SSH 协议运行。它支持 SSH 的完整安全性和身份验证功能。

先决条件任务¶

要使用这些操作符，您必须执行以下几项操作

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

按照 Google Cloud 文档中的描述，启用您项目的结算功能。

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

操作符¶

SFTP 和 Google Storage 之间的文件传输通过 SFTPToGCSOperator 操作符执行。

使用 Jinja 模板，使用 source_path、destination_path、destination_bucket、impersonation_chain 来动态定义值。

复制单个文件¶

以下操作符复制单个文件。

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 文档