Apache Airflow-DAG in Cloud Composer 2 ausführen

Cloud Composer 3 | Cloud Composer 2 | Cloud Composer 1

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

Hinweis

  1. Melden Sie sich in Ihrem Google Cloud -Konto an. Wenn Sie noch kein Google Cloud-Konto haben, 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.

    Roles required to select or create a project

    • Select a project: Selecting a project doesn't require a specific IAM role—you can select any project that you've been granted a role on.
    • Create a project: To create a project, you need the Project Creator role (roles/resourcemanager.projectCreator), which contains the resourcemanager.projects.create permission. Learn how to grant roles.

    Go to project selector

  3. Verify that billing is enabled for your Google Cloud project.

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

    Roles required to select or create a project

    • Select a project: Selecting a project doesn't require a specific IAM role—you can select any project that you've been granted a role on.
    • Create a project: To create a project, you need the Project Creator role (roles/resourcemanager.projectCreator), which contains the resourcemanager.projects.create permission. Learn how to grant roles.

    Go to project selector

  5. Verify that billing is enabled for your Google Cloud project.

  6. die Cloud Composer API aktivieren

    Rollen, die zum Aktivieren von APIs erforderlich sind

    Zum Aktivieren von APIs benötigen Sie die IAM-Rolle „Service Usage-Administrator“ (roles/serviceusage.serviceUsageAdmin), die die Berechtigung serviceusage.services.enable enthält. Informationen zum Zuweisen von Rollen.

    API aktivieren

  7. Bitten Sie Ihren Administrator, Ihnen die folgenden IAM-Rollen für Ihr Projekt zuzuweisen, um die Berechtigungen zu erhalten, die Sie zum Ausführen der Kurzanleitung benötigen,

    Weitere Informationen zum Zuweisen von Rollen finden Sie unter Zugriff auf Projekte, Ordner und Organisationen verwalten.

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

Dienstkonto für eine Umgebung erstellen

Wenn Sie eine Umgebung erstellen, geben Sie ein Dienstkonto an. Dieses Dienstkonto wird als Dienstkonto der Umgebung bezeichnet. Ihre Umgebung verwendet dieses Dienstkonto für die meisten Vorgänge.

Das Dienstkonto für Ihre Umgebung ist kein Nutzerkonto. Ein Dienstkonto ist ein spezieller Kontotyp, der nicht von einer Person, sondern von einer Anwendung oder einer VM-Instanz verwendet wird.

So erstellen Sie ein Dienstkonto für Ihre Umgebung:

  1. Erstellen Sie ein neues Dienstkonto, wie in der Dokumentation zu Identity and Access Management beschrieben.

  2. Weisen Sie ihm eine Rolle zu, wie in der Dokumentation zu Identity and Access Management beschrieben. Die erforderliche Rolle ist Composer-Worker (composer.worker).

Umgebung erstellen

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

    Zur Seite „Umgebung erstellen“

  1. Wenn dies die erste Umgebung in Ihrem Projekt ist, wird der Bereich Dem Cloud Composer-Dienstkonto die erforderlichen Berechtigungen erteilen angezeigt.

    Fügen Sie das Dienstkonto des Cloud Composer-Agents als neuen Hauptnutzer zum Dienstkonto Ihrer Umgebung hinzu und weisen Sie ihm die Rolle API-Dienst-Agent-Erweiterung von Cloud Composer v2 zu.

    Bestätigen Sie, dass Sie das gewünschte Dienstkonto für Ihre Umgebung verwenden, und klicken Sie auf Gewähren.

  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 wurde.

  6. Danach wird neben dem Umgebungsnamen ein grünes Häkchen angezeigt.

DAG-Datei erstellen

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

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

  1. Erstellt einen DAG, composer_sample_dag. Dieser DAG wird jeden Tag ausgeführt.
  2. Die Aufgabe print_dag_run_conf wird ausgeführt. Sie gibt mithilfe des Bash-Operators die Konfiguration der DAG-Ausführung 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

Jede Cloud Composer-Umgebung ist mit einem Cloud Storage-Bucket verknüpft. Airflow in Cloud Composer plant nur DAGs, die sich im Ordner /dags in diesem Bucket befinden.

Zum Planen Ihres DAG laden Sie 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 ansehen

Nachdem Sie die DAG-Datei hochgeladen haben, führt Airflow folgende Schritte aus:

  1. Die hochgeladene DAG-Datei wird geparst. Es kann einige Minuten dauern, bis der DAG für Airflow verfügbar ist.
  2. Der DAG wird der Liste der verfügbaren DAGs hinzugefügt.
  3. Der DAG wird gemäß dem Zeitplan ausgeführt, den Sie in der DAG-Datei angegeben haben.

Prüfen Sie, ob Ihr DAG ohne Fehler verarbeitet wurde und in Airflow verfügbar ist, indem Sie ihn in der DAG-UI ansehen. Die DAG-UI ist die Cloud Composer-Oberfläche zum Ansehen von DAG-Informationen in der Google Cloud Console. Cloud Composer bietet auch Zugriff auf die Airflow-UI, eine native Airflow-Web oberfläche.

  1. Warten Sie etwa fünf Minuten, damit Airflow die zuvor hochgeladene DAG-Datei verarbeiten und die erste DAG-Ausführung abschließen kann (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. Wechseln Sie zum Tab DAGs.

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

    In der Liste der DAGs wird der DAG „composer_quickstart“ mit zusätzlichen Informationen wie Status und Zeitplan angezeigt.
    Abbildung 1 Die Liste der DAGs zeigt den composer_quickstart DAG (zum Vergrößern klicken)

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 holt Airflow den Zeitplan des angegebenen DAG ein.

Der Beispiel-DAG enthält eine Aufgabe, print_dag_run_conf, die den Befehl echo in der Console ausführt. Dieser Befehl gibt Metainformationen zum DAG aus (numerische ID der DAG-Ausführung).

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

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

    In der Liste der DAG-Ausführungen wird die letzte DAG-Ausführung (mit Ausführungsdatum und Status) angezeigt.
    Abbildung 2 Die Liste der DAG-Ausführungen für den DAG „composer_quickstart“ (zum Vergrößern klicken)

  3. Die Details der DAG-Ausführung werden angezeigt und enthalten Informationen zu den einzelnen Aufgaben des Beispiel-DAG.

    Die Liste der Aufgaben mit dem Eintrag „print_dag_run_conf“ sowie deren Startzeit, Endzeit und Dauer
    Abbildung 3 Die Liste der Aufgaben, die in der DAG-Ausführung ausgeführt wurden (zum Vergrößern klicken)

  4. Im Abschnitt Logs für DAG-Ausführung sind die Logs für alle Aufgaben in der DAG-Ausführung aufgeführt. Die Ausgabe des Befehls echo ist in den Logs zu sehen.

    Logeinträge der Aufgabe, einer davon ist „Output“ und der andere enthält eine Kennung.
    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 heißen us-central1-example-environ-c1616fe8-bucket.

Nächste Schritte