Cloud Composer 1 befindet sich im Modus nach der Wartung. Google veröffentlicht keine weiteren Updates für Cloud Composer 1, einschließlich neuer Versionen von Airflow, Fehlerkorrekturen und Sicherheitsupdates. Wir empfehlen die Migration zu Cloud Composer 2.

Diese Seite wurde von der Cloud Translation API übersetzt.

Dataproc Serverless-Arbeitslasten mit Cloud Composer ausführen

Cloud Composer 3 | Cloud Composer 2 | Cloud Composer 1

Auf dieser Seite wird beschrieben, wie Sie mit Cloud Composer 2 Dataproc-Serverless-Nutzlasten aufGoogle Cloudausführen.

In den folgenden Abschnitten finden Sie Beispiele für die Verwendung von Operatoren zum Verwalten von Dataproc Serverless-Batcharbeitslasten. Sie verwenden diese Operatoren in DAGs, mit denen eine serverlose Dataproc-Spark-Batcharbeitslast erstellt, gelöscht, aufgelistet und abgerufen wird:

Erstellen Sie DAGs für Operatoren, die mit serverlosen Dataproc-Batch-Arbeitslasten funktionieren:
DAGs erstellen, die benutzerdefinierte Container und Dataproc Metastore verwenden
Konfigurieren Sie den Persistent History Server für diese DAGs.

Hinweis

Aktivieren Sie die Dataproc API:
Console

Enable the Dataproc API.
Enable the API
gcloud

Enable the Dataproc API:
```
gcloud services enable dataproc.googleapis.com
```
Wählen Sie den Speicherort für die Batcharbeitslastdatei aus. Sie haben folgende Möglichkeiten:
- Erstellen Sie einen Cloud Storage-Bucket, in dem diese Datei gespeichert wird.
- Verwenden Sie den Bucket Ihrer Umgebung. Da Sie diese Datei nicht mit Airflow synchronisieren müssen, können Sie einen separaten Unterordner außerhalb der Ordner /dags oder /data erstellen. Beispiel: /batches
- Verwenden Sie einen vorhandenen Bucket.

Dateien und Airflow-Variablen einrichten

In diesem Abschnitt wird gezeigt, wie Sie Dateien einrichten und Airflow-Variablen für diese Anleitung konfigurieren.

Dataproc Serverless Spark ML-Arbeitslastdatei in einen Bucket hochladen

In der Arbeitslast in dieser Anleitung wird ein PySpark-Script ausgeführt:

Speichern Sie ein PySpark-Script in einer lokalen Datei mit dem Namen spark-job.py. Sie können beispielsweise das Beispiel-PySpark-Script verwenden.
Laden Sie die Datei an den Speicherort hoch, den Sie unter Vorab ausgewählt haben.

Airflow-Variablen festlegen

In den folgenden Abschnitten werden Airflow-Variablen verwendet. Sie legen Werte für diese Variablen in Airflow fest, damit Ihr DAG-Code auf diese Werte zugreifen kann.

In den Beispielen in dieser Anleitung werden die folgenden Airflow-Variablen verwendet. Sie können sie je nach Beispiel nach Bedarf festlegen.

Legen Sie die folgenden Airflow-Variablen für die Verwendung in Ihrem DAG-Code fest:

project_id: Projekt-ID.
bucket_name: URI eines Buckets, in dem sich die Haupt-Python-Datei der Arbeitslast (spark-job.py) befindet. Sie haben diesen Speicherort unter Vorbereitung ausgewählt.
phs_cluster: Name des Clusters des Persistent History-Servers. Sie legen diese Variable fest, wenn Sie einen Persistent History Server erstellen.
image_name: Name und Tag des benutzerdefinierten Container-Images (image:tag). Sie legen diese Variable fest, wenn Sie ein benutzerdefiniertes Container-Image mit DataprocCreateBatchOperator verwenden.
metastore_cluster: Name des Dataproc Metastore-Dienstes. Sie legen diese Variable fest, wenn Sie den Dataproc Metastore-Dienst mit DataprocCreateBatchOperator verwenden.
region_name: Region, in der sich der Dataproc Metastore-Dienst befindet. Sie legen diese Variable fest, wenn Sie den Dataproc Metastore-Dienst mit DataprocCreateBatchOperator verwenden.

Jede Airflow-Variable über die Google Cloud Console und die Airflow-Benutzeroberfläche festlegen

Rufen Sie in der Google Cloud Console die Seite Umgebungen auf.

Zur Seite Umgebungen
Klicken Sie in der Liste der Umgebungen auf den Link Airflow für Ihre Umgebung. Die Airflow-Benutzeroberfläche wird geöffnet.
Wählen Sie in der Airflow-Benutzeroberfläche Verwaltung > Variablen aus.
Klicken Sie auf Neuen Eintrag hinzufügen.
Geben Sie den Namen der Variablen im Feld Key und den Wert für sie im Feld Val an.
Klicken Sie auf Speichern.

Persistent History Server erstellen

Verwenden Sie einen Persistent History Server (PHS), um Spark-Verlaufsdateien Ihrer Batch-Arbeitslasten aufzurufen:

Erstellen Sie einen Persistent History Server.
Achten Sie darauf, dass Sie den Namen des PHS-Clusters in der phs_cluster Airflow-Variable angegeben haben.

DataprocCreateBatchOperator

Der folgende DAG startet eine serverlose Dataproc-Batcharbeitslast.

Weitere Informationen zu DataprocCreateBatchOperator-Argumenten finden Sie im Quellcode des Operators.

Weitere Informationen zu Attributen, die du im Parameter batch von DataprocCreateBatchOperator übergeben kannst, findest du in der Beschreibung der Batch-Klasse.


"""
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

Benutzerdefiniertes Container-Image mit DataprocCreateBatchOperator verwenden

Das folgende Beispiel zeigt, wie Sie Ihre Arbeitslasten mit einem benutzerdefinierten Container-Image ausführen. Sie können einen benutzerdefinierten Container beispielsweise verwenden, um Python-Abhängigkeiten hinzuzufügen, die nicht vom Standard-Container-Image bereitgestellt werden.

So verwenden Sie ein benutzerdefiniertes Container-Image:

Erstellen Sie ein benutzerdefiniertes Container-Image und laden Sie es in Container Registry hoch.
Geben Sie das Bild in der image_name Airflow-Variablen an.
Verwenden Sie DataprocCreateBatchOperator mit Ihrem benutzerdefinierten Image:

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

Dataproc Metastore-Dienst mit DataprocCreateBatchOperator verwenden

So verwenden Sie einen Dataproc Metastore-Dienst über einen DAG:

Prüfen Sie, ob Ihr Metadatenspeicherdienst bereits gestartet wurde.

Informationen zum Starten eines Metastore-Dienstes finden Sie unter Dataproc Metastore aktivieren und deaktivieren.

Ausführliche Informationen zum Batch-Operator zum Erstellen der Konfiguration finden Sie unter PeripheralsConfig.
Geben Sie den Namen des Metastore-Dienstes in der Variablen metastore_cluster und die Region in der region_name-Airflow-Variable an, sobald er betriebsbereit ist.
Metastore-Dienst in DataprocCreateBatchOperator verwenden:

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

Mit DataprocDeleteBatchOperator können Sie einen Batch anhand der Batch-ID der Arbeitslast löschen.

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

DataprocListBatchesOperator

Mit DataprocDeleteBatchOperator werden Batches aufgelistet, die in einer bestimmten project_id und Region vorhanden sind.

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

DataprocGetBatchOperator

Mit DataprocGetBatchOperator wird eine bestimmte Batcharbeitslast abgerufen.

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