使用 Managed Airflow 的工作流程

在本文件中，您會使用下列 Google Cloud的計費元件：

  • Managed Service for Apache Spark
  • Compute Engine
  • Managed Airflow

如要根據預測用量估算費用，請使用 Pricing Calculator。

初次使用 Google Cloud 的使用者可能符合免費試用期資格。

事前準備

設定專案

  1. 登入 Google Cloud 帳戶。如果您是 Google Cloud新手，歡迎 建立帳戶，親自評估產品在實際工作環境中的成效。新客戶還能獲得價值 $300 美元的免費抵免額，可用於執行、測試及部署工作負載。

  2. In the Google Cloud console, on the project selector page, select or create a Google Cloud project.

    Roles required to select or create a project

    • Select a project: Selecting a project doesn't require a specific IAM role—you can select any project that you've been granted a role on.
    • Create a project: To create a project, you need the Project Creator role (roles/resourcemanager.projectCreator), which contains the resourcemanager.projects.create permission. Learn how to grant roles.

    Go to project selector

  3. Verify that billing is enabled for your Google Cloud project.

  4. Enable the Dataproc, Compute Engine, and Managed Airflow APIs.

    Roles required to enable APIs

    To enable APIs, you need the Service Usage Admin IAM role (roles/serviceusage.serviceUsageAdmin), which contains the serviceusage.services.enable permission. Learn how to grant roles.

    Enable the APIs

  5. 安裝 Google Cloud CLI。

  6. 若您採用的是外部識別資訊提供者 (IdP)，請先使用聯合身分登入 gcloud CLI。

  7. 執行下列指令，初始化 gcloud CLI：

    gcloud init

  8. In the Google Cloud console, on the project selector page, select or create a Google Cloud project.

    Roles required to select or create a project

    • Select a project: Selecting a project doesn't require a specific IAM role—you can select any project that you've been granted a role on.
    • Create a project: To create a project, you need the Project Creator role (roles/resourcemanager.projectCreator), which contains the resourcemanager.projects.create permission. Learn how to grant roles.

    Go to project selector

  9. Verify that billing is enabled for your Google Cloud project.

  10. Enable the Dataproc, Compute Engine, and Managed Airflow APIs.

    Roles required to enable APIs

    To enable APIs, you need the Service Usage Admin IAM role (roles/serviceusage.serviceUsageAdmin), which contains the serviceusage.services.enable permission. Learn how to grant roles.

    Enable the APIs

  11. 安裝 Google Cloud CLI。

  12. 若您採用的是外部識別資訊提供者 (IdP)，請先使用聯合身分登入 gcloud CLI。

  13. 執行下列指令，初始化 gcloud CLI：

    gcloud init

建立 Managed Service for Apache Spark 工作流程範本

請複製下列指令，並在本機終端機視窗或 Cloud Shell 執行，以建立及定義工作流程範本。

  1. 建立 sparkpi 工作流程範本。 
    gcloud dataproc workflow-templates create sparkpi \
    --region=us-central1
  2. 將 Spark 工作新增至 sparkpi 工作流程範本。「compute」step-id 標記會識別 SparkPi 工作。
    gcloud dataproc workflow-templates add-job spark \
    --workflow-template=sparkpi \
    --step-id=compute \
    --class=org.apache.spark.examples.SparkPi \
    --jars=file:///usr/lib/spark/examples/jars/spark-examples.jar \
    --region=us-central1 \
    -- 1000
  3. 使用代管的單一節點叢集執行工作流程。Managed Service for Apache Spark 會建立叢集、在叢集上執行工作流程，然後在工作流程完成時刪除叢集。
    gcloud dataproc workflow-templates set-managed-cluster sparkpi \
    --cluster-name=sparkpi \
    --single-node \
    --region=us-central1
  4. 確認工作流程範本建立完成。

    控制台

    在 Google Cloud 控制台的 Managed Service for Apache Spark「Workflows」(工作流程) 頁面中，按一下 sparkpi 名稱，開啟「Workflow template details」(工作流程範本詳細資料) 頁面。按一下工作流程範本名稱，確認 sparkpi 範本屬性。

    gcloud 指令

    執行下列指令：

    gcloud dataproc workflow-templates describe sparkpi --region=us-central1

建立 DAG 並上傳至 Cloud Storage

  1. 建立或使用現有的代管 Airflow 環境。
  2. 設定環境變數。

    Airflow UI

    1. 依序按一下工具列中的「管理」>「變數」。
    2. 按一下「建立」。
    3. 輸入下列資訊：
      • 按鍵：project_id
      • Val：PROJECT_ID - 您的 Google Cloud 專案 ID
    4. 按一下 [儲存]。

    gcloud 指令

    輸入下列指令：

    • ENVIRONMENT 是 Managed Airflow 環境的名稱
    • LOCATION 是 Managed Airflow 環境所在的地區
    • PROJECT_ID 是包含 Managed Airflow 環境的專案 ID
        gcloud composer environments run ENVIRONMENT --location LOCATION variables set -- project_id PROJECT_ID
  3. 在本機將下列 DAG 程式碼複製到名為「composer-dataproc-dag.py」的檔案中，該檔案會使用 DataprocInstantiateWorkflowTemplateOperator。

    Airflow 2

    
"""Example Airflow DAG that kicks off a Cloud Dataproc Template that runs a
Spark Pi Job.

This DAG relies on an Airflow variable
https://airflow.apache.org/docs/apache-airflow/stable/concepts/variables.html
* project_id - Google Cloud Project ID to use for the Cloud Dataproc Template.
"""

import datetime

from airflow import models
from airflow.providers.google.cloud.operators.dataproc import (
    DataprocInstantiateWorkflowTemplateOperator,
)
from airflow.utils.dates import days_ago

project_id = "{{var.value.project_id}}"


default_args = {
    # Tell airflow to start one day ago, so that it runs as soon as you upload it
    "start_date": days_ago(1),
    "project_id": project_id,
}

# Define a DAG (directed acyclic graph) of tasks.
# Any task you create within the context manager is automatically added to the
# DAG object.
with models.DAG(
    # The id you will see in the DAG airflow page
    "dataproc_workflow_dag",
    default_args=default_args,
    # The interval with which to schedule the DAG
    schedule_interval=datetime.timedelta(days=1),  # Override to match your needs
) as dag:
    start_template_job = DataprocInstantiateWorkflowTemplateOperator(
        # The task id of your job
        task_id="dataproc_workflow_dag",
        # The template id of your workflow
        template_id="sparkpi",
        project_id=project_id,
        # The region for the template
        region="us-central1",
    )

    Airflow 1

    
"""Example Airflow DAG that kicks off a Cloud Dataproc Template that runs a
Spark Pi Job.

This DAG relies on an Airflow variable
https://airflow.apache.org/docs/apache-airflow/stable/concepts/variables.html
* project_id - Google Cloud Project ID to use for the Cloud Dataproc Template.
"""

import datetime

from airflow import models
from airflow.contrib.operators import dataproc_operator
from airflow.utils.dates import days_ago

project_id = "{{var.value.project_id}}"


default_args = {
    # Tell airflow to start one day ago, so that it runs as soon as you upload it
    "start_date": days_ago(1),
    "project_id": project_id,
}

# Define a DAG (directed acyclic graph) of tasks.
# Any task you create within the context manager is automatically added to the
# DAG object.
with models.DAG(
    # The id you will see in the DAG airflow page
    "dataproc_workflow_dag",
    default_args=default_args,
    # The interval with which to schedule the DAG
    schedule_interval=datetime.timedelta(days=1),  # Override to match your needs
) as dag:
    start_template_job = dataproc_operator.DataprocWorkflowTemplateInstantiateOperator(
        # The task id of your job
        task_id="dataproc_workflow_dag",
        # The template id of your workflow
        template_id="sparkpi",
        project_id=project_id,
        # The region for the template
        # For more info on regions where Dataflow is available see:
        # https://cloud.google.com/dataflow/docs/resources/locations
        region="us-central1",
    )
  4. 將 DAG 上傳到 Cloud Storage 中的環境資料夾。上傳成功後，請按一下 Managed Airflow 環境頁面上的「DAGs Folder」連結。

查看工作狀態

Airflow UI

  1. 開啟 Airflow 網頁介面。
  2. 在 DAG 頁面中，按一下 DAG 名稱 (例如 dataproc_workflow_dag)。
  3. 在 DAG 詳細資料頁面中，按一下「Graph View」(圖表檢視)。
  4. 查看狀態：
    • 失敗：工作周圍有紅色方框。 你也可以將游標懸停在工作上，然後尋找「狀態：失敗」。工作周圍有紅色方塊，表示工作失敗
    • 成功：工作周圍會顯示綠色方塊。 您也可以將指標懸停在工作上，檢查是否顯示「State: Success」。 工作周圍有綠色方塊，表示工作已成功完成

控制台

按一下「工作流程」分頁標籤，即可查看工作流程狀態。

gcloud 指令

gcloud dataproc operations list \
    --region=us-central1 \
    --filter="labels.goog-dataproc-workflow-template-id=sparkpi"

清除所用資源

如要避免系統向您的 Google Cloud 帳戶收費，請刪除本教學課程所用的資源：

  1. 刪除 Managed Airflow 環境。

  2. 刪除工作流程範本。

後續步驟