使用 Managed Airflow 執行 Managed Service for Apache Spark 工作負載

Managed Airflow (第 3 代) | Managed Airflow (第 2 代) | Managed Airflow (舊版第 1 代)

本頁面說明如何使用 Managed Airflow (第 2 代)，在Google Cloud執行 Managed Service for Apache Spark 工作負載。

下列各節的範例說明如何使用運算子管理 Managed Service for Apache Spark 批次工作負載。您可以在 DAG 中使用這些運算子，建立、刪除、列出及取得 Managed Service for Apache Spark 批次工作負載：

• 為與 Managed Service for Apache Spark 批次工作負載搭配使用的運算子建立 DAG：

  • DataprocCreateBatchOperator
  • DataprocDeleteBatchOperator
  • DataprocListBatchesOperator
  • DataprocGetBatchOperator

• 建立使用自訂容器和 Dataproc Metastore 的 DAG。

• 為這些 DAG 設定永久記錄伺服器。

事前準備

1. 啟用 Dataproc API：

   控制台

   如果尚未啟用 Managed Service for Apache Spark API，請先啟用。

   啟用 API 所需的角色

   您必須具備 serviceusage.services.enable 權限，才能啟用 API。如果您建立了專案，可能已透過「擁有者」角色 (roles/owner) 取得這項權限。否則，您可以透過「Service Usage 管理員」角色 (roles/serviceusage.serviceUsageAdmin) 取得這項權限。瞭解如何授予角色。

   啟用 API

   gcloud

   啟用 Managed Service for Apache Spark API (如果尚未啟用)：

   啟用 API 所需的角色

   您必須具備 serviceusage.services.enable 權限，才能啟用 API。如果您建立了專案，可能已透過「擁有者」角色 (roles/owner) 取得這項權限。否則，您可以透過「Service Usage 管理員」角色 (roles/serviceusage.serviceUsageAdmin) 取得這項權限。瞭解如何授予角色。

   gcloud services enable dataproc.googleapis.com

2. 選取批次工作負載檔案的位置。您可以使用下列任一選項：

   • 建立 Cloud Storage bucket，用於儲存這個檔案。
   • 使用環境的 bucket。由於您不需要將這個檔案與 Airflow 同步，因此可以在 /dags 或 /data 資料夾外建立獨立的子資料夾。例如：/batches。
   • 使用現有 bucket。

設定檔案和 Airflow 變數

本節說明如何設定本教學課程的檔案和 Airflow 變數。

將 Managed Service for Apache Spark 機器學習工作負載檔案上傳至 bucket

本教學課程中的工作負載會執行 pyspark 指令碼：

1. 將任何 pyspark 指令碼儲存至名為 spark-job.py 的本機檔案。例如，您可以使用範例 pyspark 指令碼。

2. 將檔案上傳到您在「事前準備」中選取的位置。

設定 Airflow 變數

以下各節的範例會使用 Airflow 變數。您可以在 Airflow 中設定這些變數的值，然後 DAG 程式碼就能存取這些值。

本教學課程中的範例使用下列 Airflow 變數。您可以根據使用的範例，視需要設定這些變數。

設定下列 Airflow 變數，以便在 DAG 程式碼中使用：

• project_id：專案 ID。
• bucket_name：工作負載主要 Python 檔案 (spark-job.py) 所在 bucket 的 URI。您在「事前準備」中選取了這個位置。
• phs_cluster：永久記錄伺服器叢集名稱。您會在建立永久記錄伺服器時設定這個變數。
• image_name：自訂容器映像檔的名稱和標記 (image:tag)。您在使用 DataprocCreateBatchOperator 自訂容器映像檔時，會設定這個變數。
• metastore_cluster：Dataproc Metastore 服務名稱。當您搭配 DataprocCreateBatchOperator 使用 Dataproc Metastore 服務時，請設定這個變數。
• region_name：Dataproc Metastore 服務所在的區域。當您搭配 DataprocCreateBatchOperator 使用 Dataproc Metastore 服務時，請設定這個變數。

使用 Google Cloud 控制台和 Airflow UI 設定各個 Airflow 變數

1. 前往 Google Cloud 控制台的「Environments」(環境) 頁面。

   前往「Environments」(環境)

2. 在環境清單中，按一下環境的「Airflow」Airflow連結。Airflow UI 隨即開啟。

3. 在 Airflow UI 中，依序選取「Admin」>「Variables」。

4. 按一下「Add a new record」。

5. 在「Key」欄位中指定變數名稱，並在「Val」欄位中設定變數值。

6. 按一下 [儲存]。

建立永久記錄伺服器

使用永久記錄伺服器 (PHS) 查看批次工作負載的 Spark 記錄檔：

1. 建立永久記錄伺服器。
2. 請確認您已在 phs_cluster Airflow 變數中指定 PHS 叢集的名稱。

DataprocCreateBatchOperator

下列 DAG 會啟動 Managed Service for Apache Spark 批次工作負載。

如要進一步瞭解 DataprocCreateBatchOperator 引數，請參閱運算子的原始碼。

如要進一步瞭解可在 DataprocCreateBatchOperator 的 batch 參數中傳遞的屬性，請參閱批次類別說明。

"""
Examples below show how to use operators for managing Dataproc Serverless batch workloads.
 You use these operators in DAGs that create, delete, list, and get a Dataproc Serverless Spark batch workload.
https://airflow.apache.org/docs/apache-airflow/stable/concepts/variables.html
* project_id is the Google Cloud Project ID to use for the Cloud Dataproc Serverless.
* bucket_name is the URI of a bucket where the main python file of the workload (spark-job.py) is located.
* phs_cluster is the Persistent History Server cluster name.
* image_name is the name and tag of the custom container image (image:tag).
* metastore_cluster is the Dataproc Metastore service name.
* region_name is the region where the Dataproc Metastore service is located.
"""

import datetime

from airflow import models
from airflow.providers.google.cloud.operators.dataproc import (
    DataprocCreateBatchOperator,
    DataprocDeleteBatchOperator,
    DataprocGetBatchOperator,
    DataprocListBatchesOperator,
)
from airflow.utils.dates import days_ago

PROJECT_ID = "{{ var.value.project_id }}"
REGION = "{{ var.value.region_name}}"
BUCKET = "{{ var.value.bucket_name }}"
PHS_CLUSTER = "{{ var.value.phs_cluster }}"
METASTORE_CLUSTER = "{{var.value.metastore_cluster}}"
DOCKER_IMAGE = "{{var.value.image_name}}"

PYTHON_FILE_LOCATION = "gs://{{var.value.bucket_name }}/spark-job.py"
# for e.g.  "gs//my-bucket/spark-job.py"
# Start a single node Dataproc Cluster for viewing Persistent History of Spark jobs
PHS_CLUSTER_PATH = "projects/{{ var.value.project_id }}/regions/{{ var.value.region_name}}/clusters/{{ var.value.phs_cluster }}"
# for e.g. projects/my-project/regions/my-region/clusters/my-cluster"
SPARK_BIGQUERY_JAR_FILE = "gs://spark-lib/bigquery/spark-bigquery-latest_2.12.jar"
# use this for those pyspark jobs that need a spark-bigquery connector
# https://cloud.google.com/dataproc/docs/tutorials/bigquery-connector-spark-example
# Start a Dataproc MetaStore Cluster
METASTORE_SERVICE_LOCATION = "projects/{{var.value.project_id}}/locations/{{var.value.region_name}}/services/{{var.value.metastore_cluster }}"
# for e.g. projects/my-project/locations/my-region/services/my-cluster
CUSTOM_CONTAINER = "us.gcr.io/{{var.value.project_id}}/{{ var.value.image_name}}"
# for e.g. "us.gcr.io/my-project/quickstart-image",

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,
    "region": REGION,
}
with models.DAG(
    "dataproc_batch_operators",  # The id you will see in the DAG airflow page
    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:
    create_batch = DataprocCreateBatchOperator(
        task_id="batch_create",
        batch={
            "pyspark_batch": {
                "main_python_file_uri": PYTHON_FILE_LOCATION,
                "jar_file_uris": [SPARK_BIGQUERY_JAR_FILE],
            },
            "environment_config": {
                "peripherals_config": {
                    "spark_history_server_config": {
                        "dataproc_cluster": PHS_CLUSTER_PATH,
                    },
                },
            },
        },
        batch_id="batch-create-phs",
    )
    list_batches = DataprocListBatchesOperator(
        task_id="list-all-batches",
    )

    get_batch = DataprocGetBatchOperator(
        task_id="get_batch",
        batch_id="batch-create-phs",
    )
    delete_batch = DataprocDeleteBatchOperator(
        task_id="delete_batch",
        batch_id="batch-create-phs",
    )
    create_batch >> list_batches >> get_batch >> delete_batch

搭配 DataprocCreateBatchOperator 使用自訂容器映像檔

以下範例說明如何使用自訂容器映像檔執行工作負載。舉例來說，您可以使用自訂容器，新增預設容器映像檔未提供的 Python 依附元件。

注意： 如要執行 DAG，透過自訂 Docker 容器執行 Managed Service for Apache Spark，Managed Airflow 環境必須使用 Managed Airflow 2.0.6 和 Airflow 2.2.3 以上版本。

如要使用自訂容器映像檔，請按照下列步驟操作：

1. 建立自訂容器映像檔並上傳至 Container Registry。

2. 在 image_name Airflow 變數中指定映像檔。

3. 搭配自訂映像檔使用 DataprocCreateBatchOperator：

create_batch_with_custom_container = DataprocCreateBatchOperator(
    task_id="dataproc_custom_container",
    batch={
        "pyspark_batch": {
            "main_python_file_uri": PYTHON_FILE_LOCATION,
            "jar_file_uris": [SPARK_BIGQUERY_JAR_FILE],
        },
        "environment_config": {
            "peripherals_config": {
                "spark_history_server_config": {
                    "dataproc_cluster": PHS_CLUSTER_PATH,
                },
            },
        },
        "runtime_config": {
            "container_image": CUSTOM_CONTAINER,
        },
    },
    batch_id="batch-custom-container",
)
get_batch_custom = DataprocGetBatchOperator(
    task_id="get_batch_custom",
    batch_id="batch-custom-container",
)
delete_batch_custom = DataprocDeleteBatchOperator(
    task_id="delete_batch_custom",
    batch_id="batch-custom-container",
)
create_batch_with_custom_container >> get_batch_custom >> delete_batch_custom

搭配 DataprocCreateBatchOperator 使用 Dataproc Metastore 服務

如要透過 DAG 使用 Dataproc Metastore 服務，請按照下列步驟操作：

1. 確認 metastore 服務已啟動。

   如要瞭解如何啟動 metastore 服務，請參閱「啟用及停用 Dataproc Metastore」。

   如要進一步瞭解用於建立設定的 Batch 運算子，請參閱 PeripheralsConfig。

2. metastore 服務啟動並運作後，請在 metastore_cluster 變數中指定其名稱，並在 region_name Airflow 變數中指定其區域。

3. 在 DataprocCreateBatchOperator 中使用 metastore 服務：

create_batch_with_metastore = DataprocCreateBatchOperator(
    task_id="dataproc_metastore",
    batch={
        "pyspark_batch": {
            "main_python_file_uri": PYTHON_FILE_LOCATION,
            "jar_file_uris": [SPARK_BIGQUERY_JAR_FILE],
        },
        "environment_config": {
            "peripherals_config": {
                "metastore_service": METASTORE_SERVICE_LOCATION,
                "spark_history_server_config": {
                    "dataproc_cluster": PHS_CLUSTER_PATH,
                },
            },
        },
    },
    batch_id="dataproc-metastore",
)
get_batch_metastore = DataprocGetBatchOperator(
    task_id="get_batch_metatstore",
    batch_id="dataproc-metastore",
)
delete_batch_metastore = DataprocDeleteBatchOperator(
    task_id="delete_batch_metastore",
    batch_id="dataproc-metastore",
)

create_batch_with_metastore >> get_batch_metastore >> delete_batch_metastore

DataprocDeleteBatchOperator

您可以使用 DataprocDeleteBatchOperator，根據工作負載的批次 ID 刪除批次。

delete_batch = DataprocDeleteBatchOperator(
    task_id="delete_batch",
    batch_id="batch-create-phs",
)

DataprocListBatchesOperator

DataprocListBatchesOperator 會列出指定 project_id 和區域中的批次。

list_batches = DataprocListBatchesOperator(
    task_id="list-all-batches",
)

DataprocGetBatchOperator

DataprocGetBatchOperator 會擷取特定批次工作負載。

get_batch = DataprocGetBatchOperator(
    task_id="get_batch",
    batch_id="batch-create-phs",
)

後續步驟

• Managed Service for Apache Spark 總覽
• 使用 Cloud Run 函式觸發 DAG