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