Apache Airflow-DAG in Cloud Composer 3 ausführen

Apache Airflow-DAG in Cloud Composer 3 ausführen

Cloud Composer 1 | Cloud Composer 2 | Cloud Composer 3

In dieser Kurzanleitung erfahren Sie, wie Sie eine Cloud Composer-Umgebung erstellen und einen Apache Airflow-DAG in Cloud Composer 3 ausführen.

• Wenn Sie Airflow noch nicht kennen, finden Sie in der Anleitung zu Airflow-Konzepten in der Apache Airflow-Dokumentation weitere Informationen zu Airflow-Konzepten, -Objekten und ihrer Verwendung.

• Wenn Sie stattdessen die Google Cloud CLI verwenden möchten, lesen Sie Apache Airflow-DAG in Cloud Composer (Google Cloud CLI) ausführen.

• Wie Sie eine Umgebung mit Terraform erstellen, erfahren Sie unter Umgebungen erstellen (Terraform).

Hinweise

1. Melden Sie sich bei Ihrem Google Cloud-Konto an. Wenn Sie mit Google Cloud noch nicht vertraut sind, erstellen Sie ein Konto, um die Leistungsfähigkeit unserer Produkte in der Praxis sehen und bewerten zu können. Neukunden erhalten außerdem ein Guthaben von 300 $, um Arbeitslasten auszuführen, zu testen und bereitzustellen.

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

   Note: If you don't plan to keep the resources that you create in this procedure, create a project instead of selecting an existing project. After you finish these steps, you can delete the project, removing all resources associated with the project.

   Go to project selector

3. Die Abrechnung für das Google Cloud-Projekt muss aktiviert sein.

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

   Note: If you don't plan to keep the resources that you create in this procedure, create a project instead of selecting an existing project. After you finish these steps, you can delete the project, removing all resources associated with the project.

   Go to project selector

5. Die Abrechnung für das Google Cloud-Projekt muss aktiviert sein.

6. Cloud Composer API aktivieren.

   Aktivieren Sie die API

7. Bitten Sie Ihren Administrator, Ihnen die folgenden IAM-Rollen für Ihr Projekt zu gewähren, damit Sie die für diese Kurzanleitung erforderlichen Berechtigungen erhalten:

   • So erstellen und verwalten Sie die Cloud Composer-Umgebung:
     • Administrator für Umgebung und Storage-Objekte (roles/composer.environmentAndStorageObjectAdmin)
     • Dienstkontonutzer (roles/iam.serviceAccountUser)
   • So rufen Sie Logs auf: Loganzeige (roles/logging.viewer)

   Weitere Informationen zum Zuweisen von Rollen finden Sie unter Zugriff verwalten.

   Möglicherweise können Sie die erforderlichen Berechtigungen auch über benutzerdefinierte Rollen oder andere vordefinierte Rollen erhalten.

Umgebung erstellen

1. Rufen Sie in der Google Cloud Console die Seite Umgebung erstellen auf.

   Zur Seite „Umgebung erstellen“

2. Geben Sie im Feld Name example-environment ein.

3. Wählen Sie in der Drop-down-Liste Standort eine Region für die Cloud Composer-Umgebung aus. In dieser Anleitung wird die Region us-central1 verwendet.

4. Übernehmen Sie für die anderen Optionen der Umgebungskonfiguration die angegebenen Standardeinstellungen.

5. Klicken Sie auf Erstellen und warten Sie, bis die Umgebung erstellt ist.

   Hinweis: Es dauert etwa 25 Minuten, bis Cloud Composer die Umgebung erstellt hat.

6. Wenn der Vorgang abgeschlossen ist, wird neben dem Umgebungsnamen ein grünes Häkchen angezeigt.

DAG-Datei erstellen

Ein Airflow-DAG ist eine Sammlung organisierter Aufgaben, die Sie planen und ausführen möchten. DAGs werden in Standard-Python-Dateien definiert.

In dieser Anleitung wird ein Beispiel für einen Airflow-DAG verwendet, der in der Datei quickstart.py definiert ist. Der Python-Code in dieser Datei führt Folgendes aus:

1. Erstellt den DAG composer_sample_dag. Dieser DAG wird täglich ausgeführt.
2. Führt die Aufgabe print_dag_run_conf aus. Die Aufgabe gibt die Konfiguration der DAG-Ausführung mithilfe des bash-Operators aus.

Speichern Sie eine Kopie der Datei quickstart.py auf Ihrem lokalen Computer:

import datetime

from airflow import models
from airflow.operators import bash

# If you are running Airflow in more than one time zone
# see https://airflow.apache.org/docs/apache-airflow/stable/timezone.html
# for best practices
YESTERDAY = datetime.datetime.now() - datetime.timedelta(days=1)

default_args = {
    "owner": "Composer Example",
    "depends_on_past": False,
    "email": [""],
    "email_on_failure": False,
    "email_on_retry": False,
    "retries": 1,
    "retry_delay": datetime.timedelta(minutes=5),
    "start_date": YESTERDAY,
}

with models.DAG(
    "composer_quickstart",
    catchup=False,
    default_args=default_args,
    schedule_interval=datetime.timedelta(days=1),
) as dag:
    # Print the dag_run id from the Airflow logs
    print_dag_run_conf = bash.BashOperator(
        task_id="print_dag_run_conf", bash_command="echo {{ dag_run.id }}"
    )

DAG-Datei in den Bucket Ihrer Umgebung hochladen

Jeder Cloud Composer-Umgebung ist ein Cloud Storage-Bucket zugeordnet. Airflow in Cloud Composer plant nur DAGs, die sich im Ordner /dags in diesem Bucket befinden.

Laden Sie zum Planen Ihres DAG quickstart.py von Ihrem lokalen Computer in den Ordner /dags Ihrer Umgebung hoch:

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

   Zur Seite Umgebungen“

2. Klicken Sie in der Liste der Umgebungen auf den Namen Ihrer Umgebung: example-environment. Die Seite Umgebungsdetails wird geöffnet.

3. Klicken Sie auf DAGs-Ordner öffnen. Die Seite Bucket-Details wird geöffnet.

4. Klicken Sie auf Dateien hochladen und wählen Sie dann Ihre Kopie von quickstart.py aus.

5. Klicken Sie zum Hochladen der Datei auf Öffnen.

DAG aufrufen

Nach dem Hochladen der DAG-Datei führt Airflow die folgenden Schritte aus:

1. Analysiert die von Ihnen hochgeladene DAG-Datei. Es kann einige Minuten dauern, bis der DAG für Airflow verfügbar ist.
2. Fügt den DAG der Liste der verfügbaren DAGs hinzu.
3. Führt den DAG gemäß dem Zeitplan aus, den Sie in der DAG-Datei angegeben haben.

Prüfen Sie, ob der DAG fehlerfrei verarbeitet wird und in Airflow verfügbar ist. Rufen Sie ihn dazu in der DAG-UI auf. Die DAG-UI ist die Cloud Composer-Oberfläche zum Aufrufen von DAG-Informationen in der Google Cloud Console. Cloud Composer bietet außerdem Zugriff auf die Airflow-UI, eine native Airflow-Weboberfläche.

1. Warten Sie etwa fünf Minuten, damit Airflow die zuvor hochgeladene DAG-Datei verarbeitet und die erste DAG-Ausführung abgeschlossen hat (wird später erläutert).

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

   Zur Seite Umgebungen“

3. Klicken Sie in der Liste der Umgebungen auf den Namen Ihrer Umgebung: example-environment. Die Seite Umgebungsdetails wird geöffnet.

4. Rufen Sie den Tab DAGs auf.

5. Prüfen Sie, ob der DAG composer_quickstart in der Liste der DAGs vorhanden ist.

   

   Abbildung 1. In der Liste der DAGs wird der DAG composer_quickstart angezeigt (zum Vergrößern klicken).

   Hinweis: Der DAG airflow_monitoring wird von Cloud Composer verwendet, um den Zustand Ihrer Umgebung zu überwachen.

Details zur DAG-Ausführung ansehen

Eine einzelne Ausführung eines DAG wird als DAG-Ausführung bezeichnet. Airflow führt sofort eine DAG-Ausführung für den Beispiel-DAG aus, da das Startdatum in der DAG-Datei auf gestern festgelegt ist. Auf diese Weise entspricht Airflow dem Zeitplan des angegebenen DAG.

Der Beispiel-DAG enthält die Aufgabe print_dag_run_conf, mit der der Befehl echo in der Console ausgeführt wird. Dieser Befehl gibt Metainformationen über den DAG (die numerische Kennung der DAG-Ausführung) aus.

1. Klicken Sie auf dem Tab DAGs auf composer_quickstart. Der Tab Runs (Ausführungen) für den DAG wird geöffnet.

2. Klicken Sie in der Liste der DAG-Ausführungen auf den ersten Eintrag.

   

   Abbildung 2. Die Liste der DAG-Ausführungen für den DAG „composer_quickstart“ (zum Vergrößern klicken)

3. Es werden DAG-Ausführungsdetails mit Details zu den einzelnen Aufgaben des Beispiel-DAG angezeigt.

   

   Abbildung 3. Die Liste der Aufgaben, die bei der DAG-Ausführung ausgeführt wurden (zum Vergrößern klicken)

4. Im Abschnitt Logs für DAG-Ausführung werden Logs für alle Aufgaben in der DAG-Ausführung aufgeführt. Die Ausgabe des Befehls echo können Sie den Logs entnehmen.

   

   Abbildung 4. Logs der Aufgabe „print_dag_run_conf“ (zum Vergrößern klicken)

Bereinigen

Mit den folgenden Schritten vermeiden Sie, dass Ihrem Google Cloud-Konto die auf dieser Seite verwendeten Ressourcen in Rechnung gestellt werden.

Löschen Sie die in dieser Anleitung verwendeten Ressourcen:

1. Löschen Sie die Cloud Composer-Umgebung.

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

      Zur Seite Umgebungen“

   2. Wählen Sie example-environment aus und klicken Sie auf Löschen.

   3. Warten Sie, bis die Umgebung gelöscht ist.

2. Löschen Sie den Bucket Ihrer Umgebung. Durch das Löschen der Cloud Composer-Umgebung wird dessen Bucket nicht gelöscht.

   1. Rufen Sie in der Google Cloud Console die Seite Storage > Browser auf.

      Öffnen Sie "Storage" > "Browser".

   2. Wählen Sie den Bucket der Umgebung aus und klicken Sie auf Löschen. Dieser Bucket kann beispielsweise us-central1-example-environ-c1616fe8-bucket heißen.

3. Löschen Sie den nichtflüchtigen Speicher der Redis-Warteschlange Ihrer Umgebung. Durch das Löschen der Cloud Composer-Umgebung wird der nichtflüchtige Speicher nicht gelöscht.

   1. Wechseln Sie in der Google Cloud Console zu Compute Engine > Laufwerke.

      Zur Seite „Laufwerke“

   2. Wählen Sie den nichtflüchtigen Speicher der Redis-Warteschlange der Umgebung aus und klicken Sie auf Löschen.

      Dieses Laufwerk kann beispielsweise den Namen gke-us-central1-exampl-pvc-b12055b6-c92c-43ff-9de9-10f2cc6fc0ee haben. Laufwerke für Cloud Composer 1 haben immer den Typ Standard persistent disk und eine Größe von 2 GB.

Nächste Schritte

• DAGs hinzufügen und aktualisieren
• DAGs schreiben
• Airflow-Konfiguration überschreiben
• Auf die Airflow-UI zugreifen