Airflow-DAGs planen und auslösen

Cloud Composer 1 | Cloud Composer 2 | Cloud Composer 3

Vorschau — Cloud Composer 3

Diese Funktion unterliegt den „Nutzungsbedingungen für Pre-GA-Angebote“ im Abschnitt „Allgemeine Nutzungsbedingungen“ der dienstspezifischen Nutzungsbedingungen. Pre-GA-Funktionen sind „wie besehen“ verfügbar und werden möglicherweise nur eingeschränkt unterstützt. Weitere Informationen finden Sie in den Beschreibungen der Startphase.

Auf dieser Seite wird erläutert, wie die Planung und das Auslösen von DAGs in Airflow funktionieren, wie Sie einen Zeitplan für einen DAG definieren und wie Sie einen DAG manuell auslösen oder pausieren.

Airflow-DAGs in Cloud Composer

Airflow-DAGs in Cloud Composer werden in einer oder mehreren Cloud Composer-Umgebungen in Ihrem Projekt ausgeführt. Sie laden Quelldateien Ihrer Airflow-DAGs in einen Cloud Storage-Bucket hoch, der mit einer Umgebung verknüpft ist. Die Airflow-Instanz der Umgebung parset dann diese Dateien und plant DAG-Ausführungen gemäß dem Zeitplan der einzelnen DAGs. Während einer DAG-Ausführung plant und führt Airflow einzelne Aufgaben, die einen DAG in einer vom DAG definierten Reihenfolge bilden.

Weitere Informationen zu den Hauptkonzepten von Airflow wie Airflow-DAGs, DAG-Ausführungen, Aufgaben oder Operatoren finden Sie auf der Seite Grundlegende Konzepte in der Airflow-Dokumentation.

DAG-Planung in Airflow

Airflow bietet die folgenden Konzepte für den Planungsmechanismus:

Logisches Datum

Stellt ein Datum dar, für das eine bestimmte DAG-Ausführung ausgeführt wird.

Dies ist nicht das tatsächliche Datum, an dem Airflow einen DAG ausführt, sondern ein Zeitraum die eine bestimmte DAG-Ausführung verarbeiten muss. Für einen DAG, der jeden Tag um 12:00 Uhr ausgeführt werden soll, wäre das logische Datum ebenfalls 12:00 Uhr, an einem bestimmten Tag. Da die Ausführung zweimal täglich erfolgt, der letzten 12 Stunden. Zugleich muss die im Der DAG selbst verwendet möglicherweise gar nicht das logische Datum oder das Zeitintervall. Ein DAG kann beispielsweise dasselbe Skript einmal täglich ausführen, ohne den Wert des logischen Datums.

In Airflow-Versionen vor 2.2 wird dieses Datum als Ausführungsdatum bezeichnet.

Ausführungsdatum

Gibt das Datum an, an dem eine bestimmte DAG-Ausführung ausgeführt wird.

Bei einem DAG, der beispielsweise täglich um 12:00 Uhr ausgeführt werden soll, erfolgt die tatsächliche Ausführung des DAG möglicherweise um 12:05 Uhr, also einige Zeit nach Ablauf des logischen Datums.

Zeitplanintervall

Gibt an, wann und wie oft ein DAG in logischer Reihenfolge ausgeführt werden muss Daten.

Ein täglicher Zeitplan bedeutet beispielsweise, dass ein DAG einmal pro Tag ausgeführt wird und die logischen Datumsangaben für die DAG-Ausführungen 24-Stunden-Intervalle haben.

Startdatum

Gibt an, wann Airflow mit der Planung Ihres DAG beginnen soll.

Aufgaben in Ihrem DAG können individuelle Startzeiten haben oder Sie können ein einziges Startdatum für alle Aufgaben festlegen. Basierend auf dem Mindeststartdatum für Aufgaben in Ihrem DAG und dem Zeitplanintervall plant Airflow DAG-Ausführungen.

Catchup, Backfill und Wiederholungsversuche

Mechanismen zur Ausführung von DAG-Ausführungen für vergangene Zeiträume.

Catchup führt DAG-Ausführungen aus, die noch nicht ausgeführt wurden. Beispiel: wenn der DAG für einen längeren Zeitraum pausiert und dann wieder aktiviert wurde. Mit Backfill können Sie DAG-Ausführungen für einen bestimmten Zeitraum ausführen. Mit „Wiederholungen“ wird angegeben, wie oft Airflow versuchen muss, Aufgaben aus einem DAG auszuführen.

Die Planung funktioniert so:

1. Nach Ablauf des Startdatums wartet Airflow auf das nächste Auftreten des Zeitplanintervalls.

2. Airflow plant die erste DAG-Ausführung am Ende dieses Zeitplans Intervall.

   Beispiel: Ein DAG soll stündlich ausgeführt werden und das Startdatum ist die erste DAG-Ausführung um 13:00 Uhr.

Im Abschnitt Airflow-DAG planen in diesem Dokument wird Folgendes beschrieben: wie Sie die Planung für Ihre DAGs mit diesen Konzepten einrichten können. Weitere Informationen Informationen zu DAG-Ausführungen und zur Planung finden Sie unter DAG-Ausführungen finden Sie in der Airflow-Dokumentation.

Möglichkeiten zum Auslösen eines DAG

Airflow bietet die folgenden Möglichkeiten zum Auslösen eines DAG:

• Trigger nach Zeitplan starten. Airflow löst den DAG aus basierend auf dem in der DAG-Datei festgelegten Zeitplan.

• Manuell auslösen. Sie können einen DAG manuell über Google Cloud Console, Airflow-UI oder durch Ausführen eines Airflow-Befehlszeilenbefehls über die Google Cloud CLI.

• Trigger als Reaktion auf Ereignisse. Die Standardmethode zum Auslösen eines DAG in ist die Verwendung eines Sensors.

Weitere Möglichkeiten zum Auslösen von DAGs:

• Programmatische Auslösung Sie können einen DAG mit der Airflow REST API auslösen. Zum Beispiel aus einem Python-Skript.

• Programmatisch auf Ereignisse reagieren. Sie können DAGs als Reaktion auf Ereignisse mithilfe von Cloud Run-Funktionen und der Airflow REST API auslösen.

Hinweise

• Ihr Konto muss eine Rolle haben, mit der Objekte in den Umgebungs-Buckets verwaltet und DAGs aufgerufen und ausgelöst werden können. Weitere Informationen finden Sie unter Zugriffssteuerung.

Airflow-DAG planen

Sie definieren einen Zeitplan für einen DAG in der DAG-Datei. Bearbeiten Sie die Definition des DAG in auf folgende Weise:

1. Suchen und bearbeiten Sie die DAG-Datei auf Ihrem Computer. Wenn Sie den DAG nicht haben können Sie die zugehörige Kopie aus dem Bucket der Umgebung. Für einen neuen DAG alle Parameter definieren können, wenn Sie die DAG-Datei erstellen.

2. Definieren Sie im Parameter schedule_interval den Zeitplan. Sie können einen CRON-Ausdruck wie 0 0 * * * oder eine Voreinstellung wie @daily verwenden. Weitere Informationen finden Sie in der Airflow-Dokumentation unter Cron- und Zeitintervalle.

   Airflow bestimmt logische Datumsangaben für DAG-Ausführungen anhand des von Ihnen festgelegten Zeitplans.

3. Definieren Sie im Parameter start_date das Startdatum.

   Airflow bestimmt das logische Datum der ersten DAG-Ausführung .

4. Optional: Legen Sie mit dem Parameter catchup fest, ob Airflow alle vorherigen Ausführungen dieses DAGs vom Startdatum bis zum aktuellen Datum ausführen muss, die noch nicht ausgeführt wurden.

   Das logische Datum von DAG-Ausführungen, die während des Nachholvorgangs ausgeführt werden, liegt in der Vergangenheit. Das Ausführungsdatum entspricht dem Zeitpunkt, zu dem die DAG-Ausführung tatsächlich stattgefunden hat.

   Hinweis: Wenn Sie DAG-Ausführungen für ein bestimmtes Datum nachholen möchten, anstatt sie auf den neuesten Stand zu bringen, führen Sie den Befehl dags backfillAirflow CLI-Befehl mit der Google Cloud CLI aus.

5. Optional: Legen Sie mit dem Parameter retries fest, wie oft Airflow fehlgeschlagene Aufgaben noch einmal versuchen soll. Jeder DAG besteht aus einer oder mehreren einzelnen Aufgaben. Standardmäßig wird für Aufgaben in Cloud Composer zwei Wiederholungen ausgeführt Mal.

6. Laden Sie die neue Version des DAG in den Bucket.

7. Warten Sie, bis Airflow den DAG erfolgreich analysiert hat. Sie können beispielsweise prüfen, der Liste der DAGs in Ihrer Umgebung in der Google Cloud Console oder in der Airflow-UI

Die folgende Beispiel-DAG-Definition wird zweimal täglich um 00:00 und 12:00 Uhr ausgeführt. Das Startdatum ist auf den 1. Januar 2024 festgelegt. Airflow führt den Workflow nach dem Upload oder Pausieren jedoch nicht für vergangene Zeiträume aus, da die Funktion „Aufholen“ deaktiviert ist.

Der DAG enthält eine Aufgabe namens insert_query_job, die mit dem Operator BigQueryInsertJobOperator eine Zeile in eine Tabelle einfügt. Dieser Operator ist einer der BigQuery-Operatoren von Google Cloud, mit denen Sie Datensätze und Tabellen verwalten, Abfragen ausführen und Daten validieren können. Wenn eine bestimmte Ausführung dieser Aufgabe fehlschlägt, wird sie von Airflow noch viermal mit dem Standardwiederholungsintervall wiederholt. Das logische Datum für diese Wiederholungen bleibt gleich.

Die SQL-Abfrage für diese Zeile verwendet Airflow-Vorlagen zum Schreiben des logischen Datums eines DAG und Name in die Zeile ein.

import datetime

from airflow.models.dag import DAG
from airflow.providers.google.cloud.operators.bigquery import BigQueryInsertJobOperator

with DAG(
  "bq_example_scheduling_dag",
  start_date=datetime.datetime(2024, 1, 1),
  schedule_interval='0 */12 * * *',
  catchup=False
  ) as dag:

  insert_query_job = BigQueryInsertJobOperator(
    task_id="insert_query_job",
    retries=4,
    configuration={
        "query": {
            # schema: date (string), description (string)
            # example row: "20240101T120000", "DAG run: <DAG: bq_example_scheduling_dag>"
            "query": "INSERT example_dataset.example_table VALUES ('{{ ts_nodash }}', 'DAG run: {{ dag }}' )",
            "useLegacySql": False,
            "priority": "BATCH",
        }
    },
    location="us-central1"
  )

  insert_query_job

Zum Testen dieses DAG können Sie ihn manuell auslösen und Rufen Sie dann die Ausführungslogs der Aufgabe auf.

Weitere Beispiele für Planungsparameter

Die folgenden Beispiele für Planungsparameter veranschaulichen, wie die Planung mit Kombinationen von Parametern:

• Wenn start_date gleich datetime(2024, 4, 4, 16, 25) und schedule_interval gleich 30 16 * * * beginnt, erfolgt die erste DAG-Ausführung am 5. April 2024 um 16:30 Uhr.

• Wenn start_date gleich datetime(2024, 4, 4, 16, 35) und schedule_interval gleich 30 16 * * * beginnt, erfolgt die erste DAG-Ausführung am 6. April 2024 um 16:30 Uhr. Da das Startdatum nach dem Zeitplanintervall am 4. April 2024 liegt, die DAG-Ausführung nicht am 5. April 2024. Stattdessen endet das Zeitplanintervall am 5. April 2024 um 16:35 Uhr. Die nächste DAG-Ausführung wird also am folgenden Tag um 16:30 Uhr geplant.

• Wenn start_date datetime(2024, 4, 4) und der schedule_interval @daily ist, wird die erste DAG-Ausführung am 5. April 2024 um 00:00 Uhr geplant.

• Wenn start_date datetime(2024, 4, 4, 16, 30) ist und schedule_interval ist 0 * * * *, dann ist die erste DAG-Ausführung geplant für den 4. April 2024 um 18:00 Uhr. Nach dem angegebenen Datum und der angegebenen Uhrzeit plant Airflow eine DAG-Ausführung zur Minute 0 jeder Stunde. Der nächste Zeitpunkt, zu dem dies der Fall ist, ist 17:00 Uhr. Zu diesem Zeitpunkt plant Airflow eine DAG-Ausführung am Ende des Zeitplanintervalls, also um 18:00 Uhr.

DAG manuell auslösen

Wenn Sie einen Airflow-DAG manuell auslösen, führt Airflow ihn unabhängig vom in der DAG-Datei angegebenen Zeitplan einmal aus.

Console

So lösen Sie einen DAG über die Google Cloud Console aus:

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

   Zur Seite „Umgebungen“

2. Wählen Sie eine Umgebung aus, um die zugehörigen Details anzusehen.

3. Wechseln Sie auf der Seite Umgebungsdetails zum Tab DAGs.

4. Klicken Sie auf den Namen eines DAG.

5. Klicken Sie auf der Seite DAG-Details auf DAG auslösen. Eine neue DAG-Ausführung erstellt.

   Hinweis :Sie können keine Konfiguration für die DAG-Ausführung angeben, die Google Cloud Console

Airflow-UI

So lösen Sie einen DAG über die Airflow-Benutzeroberfläche aus:

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

   Zur Seite Umgebungen

2. Klicken Sie in der Spalte Airflow-Webserver auf den Link Airflow für Ihre Umgebung.

3. Melden Sie sich mit einem Google-Konto an, das über die entsprechenden Berechtigungen verfügt.

4. Klicken Sie in der Airflow-Weboberfläche auf der Seite DAGs in der Spalte Links für Ihren DAG auf die Schaltfläche Trigger DAG.

5. Optional: Geben Sie die DAG-Ausführungskonfiguration an.

6. Klickbasierter Trigger

gcloud

Führen Sie den Befehl dags trigger der Airflow-Befehlszeile aus:

  gcloud composer environments run ENVIRONMENT_NAME \
    --location LOCATION \
    dags trigger -- DAG_ID

Ersetzen Sie Folgendes:

• ENVIRONMENT_NAME: der Name Ihrer Umgebung
• LOCATION: Region, in der sich die Umgebung befindet.
• DAG_ID: der Name des DAG.

Weitere Informationen zum Ausführen von Befehlen der Airflow-Befehlszeile in Cloud Composer-Umgebungen finden Sie unter Befehle der Airflow-Befehlszeile ausführen.

Weitere Informationen zu den verfügbaren Airflow-Befehlszeilenbefehlen finden Sie in der Befehlsreferenz zu gcloud composer environments run.

Logs und Details zur DAG-Ausführung ansehen

In der Google Cloud Console haben Sie folgende Möglichkeiten:

• Status früherer DAG-Ausführungen und DAG-Details ansehen
• Detaillierte Protokolle aller DAG-Ausführungen und aller Aufgaben aus diesen DAGs ansehen.
• DAG-Statistiken ansehen

Darüber hinaus bietet Cloud Composer Zugriff auf die Airflow-UI, die eigene Weboberfläche von Airflow.

Hinweis: Die Seite Orchestrierung in BigQuery Planungsinformationen für alle Airflow-DAGs aus allen Cloud Composer 3-Umgebungen in Ihrem Projekt sowie andere Arten von in BigQuery verfügbare Assets.

DAG pausieren

Achtung :Wenn Sie Catchup für den DAG definiert haben, der auf einem Planen, plant Airflow dann alle nicht ausgeführten DAG-Ausführungen, wenn Sie die Pausierung aufheben den DAG.

Console

So pausieren Sie einen DAG über die Google Cloud Console:

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

   Zur Seite „Umgebungen“

2. Wählen Sie eine Umgebung aus, um die zugehörigen Details aufzurufen.

3. Wechseln Sie auf der Seite Umgebungsdetails zum Tab DAGs.

4. Klicken Sie auf den Namen eines DAG.

5. Klicken Sie auf der Seite DAG-Details auf DAG pausieren.

Airflow-UI

So pausieren Sie einen DAG über die Airflow-UI:

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

Zur Seite Umgebungen

1. Klicken Sie in der Spalte Airflow-Webserver auf den Link Airflow für Ihre Umgebung.

2. Melden Sie sich mit dem Google-Konto an, das über die entsprechenden Berechtigungen verfügt.

3. Klicken Sie in der Airflow-Weboberfläche auf der Seite DAGs auf die Ein/Aus-Schaltfläche neben dem Namen des DAG.

gcloud

Führen Sie den Befehl dags pause der Airflow-Befehlszeile aus:

  gcloud composer environments run ENVIRONMENT_NAME \
    --location LOCATION \
    dags pause -- DAG_ID

Ersetzen Sie Folgendes:

• ENVIRONMENT_NAME: der Name Ihrer Umgebung
• LOCATION: Region, in der sich die Umgebung befindet.
• DAG_ID: der Name des DAG.

Weitere Informationen zum Ausführen von Befehlen der Airflow-Befehlszeile in Cloud Composer-Umgebungen finden Sie unter Befehle der Airflow-Befehlszeile ausführen.

Weitere Informationen zu den verfügbaren Airflow-Befehlszeilenbefehlen finden Sie in der Befehlsreferenz zu gcloud composer environments run.

Nächste Schritte

• Airflow-DAGs schreiben
• Airflow-DAGs testen
• Fehlerbehebung bei Airflow-DAGs