Cloud Composer 1 Cloud Composer 2
Auf dieser Seite wird beschrieben, wie Sie mit Cloud Composer 2 Dataproc Serverless-Arbeitslasten in Google Cloud ausführen.
Die Beispiele in den folgenden Abschnitten zeigen die Verwendung von Operatoren zum Verwalten von Dataproc Serverless-Batcharbeitslasten. Sie verwenden diese Operatoren in DAGs, die eine Batcharbeitslast für Dataproc Serverless Spark erstellen, löschen, auflisten und abrufen:
Erstellen Sie DAGs für Operatoren, die mit Dataproc Serverless Batch-Arbeitslasten arbeiten:
Erstellen Sie DAGs, die benutzerdefinierte Container und Dataproc Metastore verwenden.
Konfigurieren Sie den Persistent History Server für diese DAGs.
Hinweise
Aktivieren Sie die Dataproc API:
Console
Dataproc API aktivieren.
gcloud
Aktivieren Sie die Dataproc API:
gcloud services enable dataproc.googleapis.com
Wählen Sie den Speicherort für die Batch-Arbeitslastdatei aus. Sie können eine der folgenden Optionen verwenden:
- 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-Arbeitsdatei in einen Bucket hochladen
Die Arbeitslast in dieser Anleitung führt ein Pyspark-Skript aus:
Speichern Sie jedes pyspark-Skript in einer lokalen Datei mit dem Namen
spark-job.py
. Sie können beispielsweise das Pyspark-Beispielskript verwenden.Laden Sie die Datei an den Speicherort hoch, den Sie unter Vorbereitung ausgewählt haben.
Airflow-Variablen festlegen
In den folgenden Abschnitten werden Airflow-Variablen verwendet. Sie legen Werte für diese Variablen in Airflow fest. Anschließend kann Ihr DAG-Code auf diese Werte zugreifen.
In den Beispielen in dieser Anleitung werden die folgenden Airflow-Variablen verwendet. Sie können sie je nach verwendetem Beispiel nach Bedarf festlegen.
Legen Sie die folgenden Airflow-Variablen für die Verwendung in Ihrem DAG-Code fest:
project_id
: Projekt-IDbucket_name
: URI eines Buckets, in dem sich die Python-Hauptdatei der Arbeitslast (spark-job.py
) befindet. Sie haben diesen Standort unter Hinweise 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.
Mit der Google Cloud Console und der Airflow-UI die einzelnen Airflow-Variablen festlegen
Rufen Sie in der Google Cloud Console die Seite Umgebungen auf.
Klicken Sie in der Liste der Umgebungen auf den Link Airflow für Ihre Umgebung. Die Airflow-UI wird geöffnet.
Wählen Sie in der Airflow-UI Admin > Variablen aus.
Klicken Sie auf Neuen Eintrag hinzufügen.
Geben Sie den Namen der Variablen im Feld Schlüssel an und legen Sie den Wert dafür im Feld Val fest.
Klicken Sie auf Speichern.
Persistent History Server erstellen
Verwenden Sie einen Persistent History Server (PHS), um Spark-Verlaufsdateien Ihrer Batcharbeitslasten anzusehen:
- Nichtflüchtigen Verlaufsserver erstellen
- Achten Sie darauf, dass Sie den Namen des PHS-Clusters in der Airflow-Variable
phs_cluster
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 Sie im Parameter batch
von DataprocCreateBatchOperator
übergeben können, finden Sie in der Beschreibung der Batch-Klasse.
Benutzerdefiniertes Container-Image mit DataprocCreateBatchOperator verwenden
Das folgende Beispiel zeigt, wie Sie ein benutzerdefiniertes Container-Image zum Ausführen Ihrer Arbeitslasten verwenden. Mit einem benutzerdefinierten Container können Sie beispielsweise Python-Abhängigkeiten hinzufü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 Image in der Airflow-Variable
image_name
an.Verwenden Sie DataprocCreateBatchOperator mit Ihrem benutzerdefinierten Image:
Dataproc Metastore-Dienst mit DataprocCreateBatchOperator verwenden
So verwenden Sie einen Dataproc Metastore-Dienst aus einem DAG:
Prüfen Sie, ob der Metastore-Dienst bereits gestartet wurde.
Informationen zum Starten eines Metastore-Dienstes finden Sie unter Dataproc Metastore aktivieren und deaktivieren.
Ausführliche Informationen zum Batchoperator zum Erstellen der Konfiguration finden Sie unter PeripheralsConfig.
Sobald der Metastore-Dienst einsatzbereit ist, geben Sie seinen Namen in der Variablen
metastore_cluster
und seine Region in der Airflow-Variableregion_name
an.Verwenden Sie den Metastore-Dienst in DataprocCreateBatchOperator:
DataprocDeleteBatchOperator
Mit DataprocDeleteBatchOperator können Sie einen Batch anhand der Batch-ID der Arbeitslast löschen.
DataprocListBatchesOperator
DataprocDeleteBatchOperator listet Batches auf, die in einer bestimmten Projekt-ID und Region vorhanden sind.
DataprocGetBatchOperator
DataprocGetBatchOperator ruft eine bestimmte Batcharbeitslast ab.