從向後移植供應商套件匯入運算子

Cloud Composer 3 | Cloud Composer 2 | Cloud Composer 1

本頁面說明 Airflow 回溯供應工具包，以及如何在 DAG 中使用這些工具包。

向後移植供應商套件是 Airflow 2 版本的運算子、轉移、感應器、掛鉤和機密，這些項目會封裝為 PyPI 模組。

為簡化說明，本文件只會著重於運算子。您可以使用向後移植套件中的所有其他內容，方法與運算子相同。

反向移植供應工具包可解決一個重要問題：您可以取得運算子、轉移、感應器和鉤子的全新功能和安全性更新，而無須將 Airflow 環境升級至較新版本。舉例來說，Airflow 2 提供 Google 的供應工具包。您可以在 Airflow 1.10.* 環境中，從向後移植版本匯入此套件的運算子。

在 Apache Airflow 推出向後移植供應商套件的概念之前，運算子是 Airflow 的必要元素。如要取得新版運算子，您需要使用較新版本的 Airflow。您必須將 Cloud Composer 環境升級至較新版本。Airflow 2 已捨棄這個模型，並推出供應工具包。Airflow 2 的供應工具包是 PyPI 模組，其中包含特定供應商 (例如 Google) 的運算子、轉移、感應器、鉤子和機密資料。如果您使用 Airflow 1.10.*，可以利用這項變更並使用回溯提供者套件的版本。

使用預先安裝的回溯套件

部分回溯端安裝包已安裝在 Cloud Composer 映像檔中。您不需要在環境中安裝這些回溯套件。只要確保在 DAG 程式碼中匯入向後移植套件的運算子，如本頁後續所述。

如要查看環境中可用的回溯端套件，請參閱環境的 Cloud Composer 映像檔中的套件清單。

環境中每個預先安裝的回溯端套件都有特定版本。如果您想使用其他版本，請更新環境，並指定所需版本。我們不建議降級預先安裝的回溯端套件。只有在發現較新版本有問題時，才安裝舊版本。您無法解除安裝預先安裝的回溯端套件，只能變更已安裝的版本。

安裝及升級回溯套件

如要安裝或升級回溯套件，請按照下列步驟操作：

  1. 在 PyPI.org 上尋找所需的回溯套件。

  2. 安裝或升級套件，就像安裝其他 PyPI 套件一樣。

  3. 視需要安裝跨供應器套件依附元件。這些是您可能需要使用回溯移植套件所有功能的額外依附元件。

    舉例來說，如要使用 apache-airflow-backport-providers-google 中的 SalesforceToGcsOperator，就需要 [salesforce] 額外項目。安裝 apache-airflow-backport-providers-google，並在「額外項目與版本」欄位中指定 [salesforce] 額外項目。

從 DAG 中的向後移植供應商套件匯入運算子

如要查看回溯移植套件中的內容清單，請前往 PyPI.org 的回溯移植套件頁面。舉例來說，apache-airflow-backport-providers-google 的頁面會列出這個套件的運算子、轉移、感應器、鉤子和機密。

向後移植供應商套件會引入新的運算子和其他內容。這兩種內容之間有差異，後文會說明。

匯入新運算子

新運算子是指 Airflow 1.10.* 中不存在的運算子。如果您嘗試匯入此類運算子，但未使用其回溯端套件，系統會傳回匯入錯誤。

如要使用回溯端套件中的新運算子，請從對應的 airflow.providers.* 套件匯入，如 PyPI.org 上此回溯端套件的頁面所述。

以下範例會從 apache-airflow-backport-providers-google 套件匯入新運算子：

from airflow.providers.google.cloud.operators.bigquery_dts import (
    BigQueryCreateDataTransferOperator,
    BigQueryDeleteDataTransferConfigOperator,
    )

匯入已移動的運算子

已移除的運算子是指 Airflow 1.10.* 中已存在的運算子。安裝向後移植運算子套件後，您可以匯入兩個不同版本的運算子。其中一個版本與 Airflow 捆綁在一起，另一個版本則是已移出的運算子。如要使用已移至新路徑的運算子，請使用新的匯入路徑匯入該運算子。

如要使用向後移植套件中的移除運算子，請從對應的 airflow.contrib.* 套件匯入，如 PyPI.org 上這個向後移植套件的頁面所述。

以下範例會從 apache-airflow-backport-providers-google 套件匯入已移出的運算子：

from airflow.contrib.operators.bigquery_operator import (
    BigQueryCreateEmptyDatasetOperator,
    BigQueryOperator,
    )

後續步驟