Apache Airflow-DAG in Managed Airflow (Gen 3) ausführen (Google Cloud CLI)

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

In dieser Kurzanleitung erfahren Sie, wie Sie eine Managed Service for Apache Airflow-Umgebung erstellen und einen Apache Airflow-DAG in Managed Airflow (Gen 3) ausführen.

Hinweis

  1. Melden Sie sich in Ihrem Google Cloud -Konto an. Wenn Sie mit Google Cloudnoch nicht vertraut sind, erstellen Sie einfach 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. Installieren Sie die Google Cloud CLI.

  3. Wenn Sie einen externen Identitätsanbieter (IdP) verwenden, müssen Sie sich zuerst mit Ihrer föderierten Identität in der gcloud CLI anmelden.

  4. Führen Sie den folgenden Befehl aus, um die gcloud CLI zu initialisieren: 

    gcloud init

  5. Google Cloud -Projekt erstellen oder auswählen.

    Erforderliche Rollen zum Auswählen oder Erstellen eines Projekts

    • Projekt auswählen: Für die Auswahl eines Projekts ist keine bestimmte IAM-Rolle erforderlich. Sie können jedes Projekt auswählen, für das Ihnen eine Rolle zugewiesen wurde.
    • Projekt erstellen: Zum Erstellen eines Projekts benötigen Sie die Rolle „Project Creator“ (roles/resourcemanager.projectCreator), die die Berechtigung resourcemanager.projects.create enthält. Informationen zum Zuweisen von Rollen

    • Erstellen Sie ein Projekt in Google Cloud :

      gcloud projects create PROJECT_ID

      Ersetzen Sie PROJECT_ID durch einen Namen für das Google Cloud -Projekt, das Sie erstellen.

    • Wählen Sie das von Ihnen erstellte Projekt Google Cloud aus:

      gcloud config set project PROJECT_ID

      Ersetzen Sie PROJECT_ID durch den Namen Ihres Projekts in Google Cloud .

  6. Prüfen Sie, ob die Abrechnung für Ihr Google Cloud -Projekt aktiviert ist.

  7. Installieren Sie die Google Cloud CLI.

  8. Wenn Sie einen externen Identitätsanbieter (IdP) verwenden, müssen Sie sich zuerst mit Ihrer föderierten Identität in der gcloud CLI anmelden.

  9. Führen Sie den folgenden Befehl aus, um die gcloud CLI zu initialisieren: 

    gcloud init

  10. Google Cloud -Projekt erstellen oder auswählen.

    Erforderliche Rollen zum Auswählen oder Erstellen eines Projekts

    • Projekt auswählen: Für die Auswahl eines Projekts ist keine bestimmte IAM-Rolle erforderlich. Sie können jedes Projekt auswählen, für das Ihnen eine Rolle zugewiesen wurde.
    • Projekt erstellen: Zum Erstellen eines Projekts benötigen Sie die Rolle „Project Creator“ (roles/resourcemanager.projectCreator), die die Berechtigung resourcemanager.projects.create enthält. Informationen zum Zuweisen von Rollen

    • Erstellen Sie ein Projekt in Google Cloud :

      gcloud projects create PROJECT_ID

      Ersetzen Sie PROJECT_ID durch einen Namen für das Google Cloud -Projekt, das Sie erstellen.

    • Wählen Sie das von Ihnen erstellte Projekt Google Cloud aus:

      gcloud config set project PROJECT_ID

      Ersetzen Sie PROJECT_ID durch den Namen Ihres Projekts in Google Cloud .

  11. Prüfen Sie, ob die Abrechnung für Ihr Google Cloud -Projekt aktiviert ist.

  12. Aktivieren Sie die Managed Airflow API, falls sie noch nicht aktiviert ist:

    Erforderliche Rollen zum Aktivieren von APIs

    Zum Aktivieren von APIs benötigen Sie die Berechtigung serviceusage.services.enable. Wenn Sie das Projekt erstellt haben, haben Sie diese Berechtigung wahrscheinlich bereits über die Inhaberrolle (roles/owner). Andernfalls können Sie diese Berechtigung über die Rolle „Service Usage-Administrator“ (roles/serviceusage.serviceUsageAdmin) erhalten. Informationen zum Zuweisen von Rollen 

    gcloud services enable composer.googleapis.com

  13. 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, um die meisten Vorgänge auszuführen.

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

Erstellen Sie eine neue Umgebung mit dem Namen example-environment in der Region us-central1 mit der neuesten Version von Managed Airflow (Gen 3).

gcloud composer environments create example-environment \
    --location us-central1 \
    --image-version composer-3-airflow-2.11.1-build.21 \
    --service-account ENVIRONMENT_SERVICE_ACCOUNT

Ersetzen Sie ENVIRONMENT_SERVICE_ACCOUNT durch das Dienstkonto für Ihre Umgebung, das Sie zuvor erstellt haben.

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 Airflow-Beispiel-DAG verwendet, der in der Datei quickstart.py definiert ist. Der Python-Code in dieser Datei führt Folgendes aus:

  1. Erstellt einen DAG, composer_sample_dag. Dieser DAG wird täglich 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 Managed Airflow-Umgebung ist mit einem Cloud Storage-Bucket verknüpft. Airflow in Managed Airflow plant nur DAGs, die sich im Ordner /dags in diesem Bucket befinden.

Wenn Sie den DAG planen möchten, laden Sie quickstart.py von Ihrem lokalen Computer in den Ordner /dags Ihrer Umgebung hoch:

Führen Sie den folgenden Befehl in dem Ordner aus, in dem sich die Datei quickstart.py befindet, um quickstart.py mit der Google Cloud CLI hochzuladen:

gcloud composer environments storage dags import \
--environment example-environment --location us-central1 \
--source quickstart.py

DAG ansehen

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

  1. Parst 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 Ihr DAG fehlerfrei verarbeitet wird und in Airflow verfügbar ist, indem Sie ihn in der DAG-UI aufrufen. Die DAG-Benutzeroberfläche ist die Managed Airflow-Oberfläche zum Aufrufen von DAG-Informationen in der Google Cloud Console. Managed Airflow bietet auch Zugriff auf die Airflow-UI, eine native Airflow-Weboberflä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. Führen Sie den folgenden Befehl in der Google Cloud CLI aus. Mit diesem Befehl wird der Airflow-Befehlszeilenbefehl dags list ausgeführt, der die DAGs in Ihrer Umgebung auflistet.

    gcloud composer environments run example-environment \
--location us-central1 \
dags list

  3. Prüfen Sie, ob der DAG composer_quickstart in der Ausgabe des Befehls aufgeführt ist.

    Beispielausgabe:

    Executing the command: [ airflow dags list ]...
Command has been started. execution_id=d49074c7-bbeb-4ee7-9b26-23124a5bafcb
Use ctrl-c to interrupt the command
dag_id              | filepath              | owner            | paused
====================+=======================+==================+=======
airflow_monitoring  | airflow_monitoring.py | airflow          | False
composer_quickstart | dag-quickstart-af2.py | Composer Example | False

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

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 Kennung der DAG-Ausführung).

Führen Sie den folgenden Befehl in der Google Cloud CLI aus. Mit diesem Befehl werden DAG-Ausführungen für den DAG composer_quickstart aufgelistet:

gcloud composer environments run example-environment \
--location us-central1 \
dags list-runs -- --dag-id composer_quickstart

Beispielausgabe:

dag_id              | run_id                                      | state   | execution_date                   | start_date                       | end_date
====================+=============================================+=========+==================================+==================================+=================================
composer_quickstart | scheduled__2024-02-17T15:38:38.969307+00:00 | success | 2024-02-17T15:38:38.969307+00:00 | 2024-02-18T15:38:39.526707+00:00 | 2024-02-18T15:38:42.020661+00:00

Die Airflow-Befehlszeile bietet keinen Befehl zum Aufrufen von Aufgabenlogs. Sie können andere Methoden verwenden, um Airflow-Aufgabenlogs aufzurufen: Managed Airflow-DAG-UI, Airflow-UI oder Cloud Logging. In diesem Leitfaden wird beschrieben, wie Sie Cloud Logging nach Logs aus einer bestimmten DAG-Ausführung abfragen.

Führen Sie den folgenden Befehl in der Google Cloud CLI aus. Mit diesem Befehl werden Logs aus Cloud Logging für eine bestimmte DAG-Ausführung des DAG composer_quickstart gelesen. Mit dem Argument --format wird die Ausgabe so formatiert, dass nur der Text des Logeintrags angezeigt wird.

gcloud logging read \
--format="value(textPayload)" \
--order=asc \
"resource.type=cloud_composer_environment \
resource.labels.location=us-central1 \
resource.labels.environment_name=example-environment \
labels.workflow=composer_quickstart \
(labels.\"execution-date\"=\"RUN_ID\")"

Ersetzen Sie:

  • RUN_ID durch den Wert run_id aus der Ausgabe des zuvor ausgeführten Befehls tasks states-for-dag-run. Beispiel: 2024-02-17T15:38:38.969307+00:00.

Beispielausgabe:

...

Starting attempt 1 of 2
Executing <Task(BashOperator): print_dag_run_conf> on 2024-02-17
15:38:38.969307+00:00
Started process 22544 to run task

...

Running command: ['/usr/bin/bash', '-c', 'echo 115746']
Output:
115746

...

Command exited with return code 0
Marking task as SUCCESS. dag_id=composer_quickstart,
task_id=print_dag_run_conf, execution_date=20240217T153838,
start_date=20240218T153841, end_date=20240218T153841
Task exited with return code 0
0 downstream tasks scheduled from follow-on schedule check

Bereinigen

Löschen Sie das Projekt von Google Cloud zusammen mit den Ressourcen, damit Ihrem Konto von Google Cloud die auf dieser Seite verwendeten Ressourcen nicht in Rechnung gestellt werden.

Löschen Sie die in dieser Anleitung verwendeten Ressourcen:

  1. Löschen Sie die Managed Airflow-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. Wenn Sie die Managed Airflow-Umgebung löschen, wird der zugehörige Bucket nicht gelöscht.

    1. Wechseln Sie in der Google Cloud Console zur Seite Storage > Browser.

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

Nächste Schritte