Menjalankan workload Managed Service for Apache Spark dengan Managed Airflow

Managed Airflow (Gen 3) | Managed Airflow (Gen 2) | Managed Airflow (Legacy Gen 1)

Halaman ini menjelaskan cara menggunakan Managed Airflow (Gen 2) untuk menjalankan Managed Service untuk Apache Spark workload di Google Cloud.

Contoh di bagian berikut menunjukkan cara menggunakan operator untuk mengelola workload batch Managed Service untuk Apache Spark. Anda menggunakan operator ini dalam DAG yang membuat, menghapus, mencantumkan, dan mendapatkan workload batch Managed Service untuk Apache Spark:

Sebelum memulai

  1. Aktifkan Dataproc API:

    Konsol

    Aktifkan Managed Service untuk Apache Spark API.

    Peran yang diperlukan untuk mengaktifkan API

    Untuk mengaktifkan API, Anda memerlukan izin serviceusage.services.enable. Jika Anda membuat project, kemungkinan Anda sudah memiliki izin ini melalui peran Pemilik (roles/owner). Jika tidak, Anda bisa mendapatkan izin ini melalui peran Admin Penggunaan Layanan (roles/serviceusage.serviceUsageAdmin). Pelajari cara memberikan peran.

    Mengaktifkan API

    gcloud

    Aktifkan Managed Service untuk Apache Spark API:

    Peran yang diperlukan untuk mengaktifkan API

    Untuk mengaktifkan API, Anda memerlukan izin serviceusage.services.enable. Jika Anda membuat project, kemungkinan Anda sudah memiliki izin ini melalui peran Pemilik (roles/owner). Jika tidak, Anda bisa mendapatkan izin ini melalui peran Admin Penggunaan Layanan (roles/serviceusage.serviceUsageAdmin). Pelajari cara memberikan peran. 

    gcloud services enable dataproc.googleapis.com

  2. Pilih lokasi untuk file workload batch Anda. Anda dapat menggunakan salah satu opsi berikut:

    • Buat bucket Cloud Storage yang menyimpan file ini.
    • Gunakan bucket lingkungan Anda. Karena Anda tidak perlu menyinkronkan file ini dengan Airflow, Anda dapat membuat subfolder terpisah di luar folder /dags atau /data. Misalnya, /batches.
    • Gunakan bucket yang ada.

Menyiapkan file dan variabel Airflow

Bagian ini menunjukkan cara menyiapkan file dan mengonfigurasi variabel Airflow untuk tutorial ini.

Mengupload file workload ML Managed Service untuk Apache Spark ke bucket

Workload dalam tutorial ini menjalankan skrip pyspark:

  1. Simpan skrip pyspark ke file lokal bernama spark-job.py. Misalnya, Anda dapat menggunakan contoh skrip pyspark.

  2. Unggah file ke lokasi yang Anda pilih di Sebelum memulai.

Menyetel variabel Airflow

Contoh di bagian berikut menggunakan variabel Airflow. Anda menetapkan nilai untuk variabel ini di Airflow, lalu kode DAG Anda dapat mengakses nilai ini.

Contoh dalam tutorial ini menggunakan variabel Airflow berikut. Anda dapat menetapkannya sesuai kebutuhan, bergantung pada contoh yang Anda gunakan.

Tetapkan variabel Airflow berikut untuk digunakan dalam kode DAG Anda:

Menggunakan Google Cloud konsol dan UI Airflow untuk menetapkan setiap variabel Airflow

  1. Di Google Cloud konsol, buka halaman Environments.

    Buka Environments

  2. Dalam daftar lingkungan, klik link Airflow untuk lingkungan Anda. UI Airflow akan terbuka.

  3. Di UI Airflow, pilih Admin > Variables.

  4. Klik Add a new record.

  5. Tentukan nama variabel di kolom Key, dan tetapkan nilainya di kolom Val.

  6. Klik Save.

Membuat Persistent History Server

Gunakan Persistent History Server (PHS) untuk melihat file histori Spark dari workload batch Anda:

  1. Buat Persistent History Server.
  2. Pastikan Anda menentukan nama cluster PHS dalam variabel Airflow phs_cluster .

DataprocCreateBatchOperator

DAG berikut memulai workload batch Managed Service untuk Apache Spark.

Untuk mengetahui informasi selengkapnya tentang argumen DataprocCreateBatchOperator, lihat kode sumber operator.

Untuk mengetahui informasi selengkapnya tentang atribut yang dapat Anda teruskan dalam batch parameter dari DataprocCreateBatchOperator, lihat deskripsi class 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

Menggunakan image container kustom dengan DataprocCreateBatchOperator

Contoh berikut menunjukkan cara menggunakan image container kustom untuk menjalankan workload Anda. Anda dapat menggunakan container kustom, misalnya, untuk menambahkan dependensi Python yang tidak disediakan oleh image container default.

Untuk menggunakan image container kustom:

  1. Buat image container kustom dan upload ke Container Registry.

  2. Tentukan image dalam image_name variabel Airflow.

  3. Gunakan DataprocCreateBatchOperator dengan image kustom Anda:

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

Menggunakan layanan Dataproc Metastore dengan DataprocCreateBatchOperator

Untuk menggunakan layanan Dataproc Metastore dari DAG:

  1. Pastikan layanan metastore Anda sudah dimulai.

    Untuk mempelajari cara memulai layanan metastore, lihat Mengaktifkan dan menonaktifkan Dataproc Metastore.

    Untuk mengetahui informasi mendetail tentang operator Batch untuk membuat konfigurasi, lihat PeripheralsConfig.

  2. Setelah layanan metastore aktif dan berjalan, tentukan namanya dalam variabel metastore_cluster dan regionnya dalam region_name variabel Airflow.

  3. Gunakan layanan metastore di DataprocCreateBatchOperator:

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

Anda dapat menggunakan DataprocDeleteBatchOperator untuk menghapus batch berdasarkan ID batch workload.

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

DataprocListBatchesOperator

DataprocDeleteBatchOperator mencantumkan batch yang ada dalam project_id dan region tertentu.

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

DataprocGetBatchOperator

DataprocGetBatchOperator mengambil satu workload batch tertentu.

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

Langkah berikutnya