在工作流程中執行 DLT 管線

您可以使用 Databricks 作業、Apache Airflow 或 Azure Data Factory，在數據處理工作流程中執行 DLT 管線。

工作

您可以在 Databricks 作業中協調多個工作，以實作數據處理工作流程。 若要在作業中包含 DLT Pipeline，請在建立作業時使用 Pipeline 任務。 如需作業 ，請參閱DLT 管線工作。

Apache Airflow

Apache Airflow 是一個開放原始碼解決方案，可用於管理和排程數據工作流程。 Airflow 將工作流程表示為操作的有向無環圖（DAG）。 您可以在 Python 檔案中定義工作流程，而 Airflow 會管理排程和執行。 如需搭配 Azure Databricks 安裝和使用 Airflow 的資訊，請參閱 使用 Apache Airflow協調 Azure Databricks 作業。

若要在 Airflow 工作流程中執行 DLT 管線，請使用 DatabricksSubmitRunOperator。

要求

要使用 Airflow 支援 DLT，您需要下列條件：

• Airflow 2.1.0 版或更新版本。
• Databricks 提供者套件 2.1.0 版或更高版本。

例

下列範例會建立 Airflow DAG，觸發具有識別碼 8279d543-063c-4d63-9926-dae38e35ce8b的 DLT 管線更新：

from airflow import DAG
from airflow.providers.databricks.operators.databricks import DatabricksSubmitRunOperator
from airflow.utils.dates import days_ago

default_args = {
  'owner': 'airflow'
}

with DAG('dlt',
         start_date=days_ago(2),
         schedule_interval="@once",
         default_args=default_args
         ) as dag:

  opr_run_now=DatabricksSubmitRunOperator(
    task_id='run_now',
    databricks_conn_id='CONNECTION_ID',
    pipeline_task={"pipeline_id": "8279d543-063c-4d63-9926-dae38e35ce8b"}
  )

請將工作區中的 CONNECTION_ID 取代為 的 Airflow 連線識別碼。

將此範例儲存在 airflow/dags 目錄中，並使用 Airflow UI 來 檢視並觸發 DAG。 使用 DLT UI 來檢視管線更新的詳細數據。

Azure Data Factory

Azure Data Factory 是雲端式 ETL 服務，可讓您協調數據整合和轉換工作流程。 Azure Data Factory 直接支援在工作流程中執行 Azure Databricks 工作，包括 筆記本、JAR 工作和 Python 腳本。 您也可以從 Azure Data Factory Web 活動呼叫 DLT API，來在工作流程中包含管線。 例如，若要從 Azure Data Factory 觸發管線更新：

1. 建立數據處理站 或開啟現有的數據處理站。

2. 建立完成時，開啟您的資料工廠頁面，然後按一下 [開啟 Azure Data Factory Studio] 磚片。 Azure Data Factory 使用者介面隨即出現。

3. 從 Azure Data Factory Studio 使用者介面的 [新增] 下拉功能表中選取 [Pipeline]，以建立新的 Azure Data Factory 管線。

4. 在 [活動] 工具箱中，展開 [一般]，然後將 [Web] 活動拖曳至管線畫布。 點選單擊 [設定] 索引標籤，然後輸入下列值：

   注意

   作為安全性最佳做法，當您使用自動化工具、系統、腳本和應用程式進行驗證時，Databricks 建議您使用屬於 服務主體的個人存取令牌， 而不是工作區使用者。 若要建立服務主體的權杖，請參閱 管理服務主體的權杖。

   • URL：https://<databricks-instance>/api/2.0/pipelines/<pipeline-id>/updates。

     取代 <get-workspace-instance>。

     將 <pipeline-id> 取代為管線標識碼。

   • 方法：從下拉功能表中選取 POST。

   • 標頭：按一下 + 新增。 在 [名稱] 文字框中，輸入 Authorization。 在 [值] 文字框中，輸入 Bearer <personal-access-token>。

     將 <personal-access-token> 取代為 Azure Databricks 個人存取權杖。

   • 正文：若要傳遞其他要求參數，請輸入包含參數的 JSON 檔案。 例如，若要啟動更新程序並重新處理管線的所有資料：{"full_refresh": "true"}。 如果沒有額外的要求參數，請輸入空白大括弧 （{}）。

若要測試 Web 活動，請在 Data Factory UI 中的管線工具列上按一下 [偵錯]。 執行的輸出和狀態，包括錯誤，會顯示在 Azure Data Factory 管線 [輸出] 標籤中。 使用 DLT UI 來檢視管線更新的詳細數據。