Cloud Composer 1 befindet sich im Nachwartungsmodus. Google veröffentlicht keine weiteren Updates für Cloud Composer 1, einschließlich neuer Versionen von Airflow, Fehlerkorrekturen und Sicherheitsupdates. Wir empfehlen, die Migration zu Cloud Composer 2 zu planen.

DAGs auslösen

Cloud Composer 1 Cloud Composer 2

Auf dieser Seite werden verschiedene Möglichkeiten zum Auslösen von DAGs in Cloud Composer-Umgebungen beschrieben.

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

Trigger nach Zeitplan starten. Wenn Sie einen DAG erstellen, geben Sie einen Zeitplan dafür an. Airflow löst den DAG anhand der angegebenen Planungsparameter automatisch aus.
Manuell auslösen. Sie können einen DAG manuell über die Airflow-UI oder durch einen Airflow-Befehlszeilenbefehl von gcloud auslösen.
Trigger als Reaktion auf Ereignisse. Als Standardmethode zum Auslösen eines DAG als Reaktion auf Ereignisse wird die Verwendung eines Sensors verwendet.

Weitere Möglichkeiten zum Auslösen von DAGs:

Programmatisch auslösen. Sie können einen DAG mit der Airflow REST API auslösen. z. B. von einem Python-Skript.
Programmatisch als Reaktion auf Ereignisse auslösen. Sie können DAGs als Reaktion auf Ereignisse mithilfe von Cloud Functions und der Airflow REST API auslösen.

DAG nach Zeitplan auslösen

So lösen Sie einen DAG nach Zeitplan aus:

Geben Sie dafür die Parameter start_date und schedule_interval in der DAG-Datei an, wie weiter unten in diesem Abschnitt beschrieben.
Laden Sie die DAG-Datei in Ihre Umgebung hoch.

Planungsparameter angeben

Wenn Sie einen DAG definieren, geben Sie im Parameter schedule_interval an, wie oft Sie den DAG ausführen möchten. Im Parameter start_date geben Sie an, wann Airflow Ihren DAG planen 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.

Die Planung funktioniert folgendermaßen: Nach Ablauf von start_date wartet Airflow auf das folgende Vorkommen von schedule_interval. Dann wird die erste DAG-Ausführung am Ende dieses Zeitplanintervalls geplant. Wenn ein DAG beispielsweise jede Stunde ausgeführt werden soll (schedule_interval ist eine Stunde) und das Startdatum um 12:00 Uhr ist, erfolgt die erste DAG-Ausführung heute um 13:00 Uhr.

Das folgende Beispiel zeigt einen DAG, der stündlich ab 15:00 Uhr am 5. April 2021 ausgeführt wird. Mit den im Beispiel verwendeten Parametern plant Airflow die erste DAG-Ausführung um 16:00 Uhr am 5. April 2021.

from datetime import datetime
from airflow import DAG
from airflow.operators.bash_operator import BashOperator

with DAG(
    dag_id='example_dag_schedule',
    # At 15:00 on 5 April, 2021
    start_date=datetime(2021, 4, 5, 15, 0),
    # At minute 0 of every hour
    schedule_interval='0 * * * *') as dag:

    # Output the current date and time
    t1 = BashOperator(
        task_id='date',
        bash_command='date',
        dag=dag)

    t1

Weitere Informationen zu den Planungsparametern finden Sie in der Airflow-Dokumentation unter DAG-Ausführungen.

Weitere Beispiele für Planungsparameter

Die folgenden Beispiele für Planungsparameter zeigen, wie die Planung mit verschiedenen Parameterkombinationen funktioniert:

Wenn start_date den Wert datetime(2021, 4, 4, 16, 25) und schedule_interval den Wert 30 16 * * * hat, erfolgt die erste DAG-Ausführung um 16:30 Uhr am 5. April 2021.
Wenn start_date den Wert datetime(2021, 4, 4, 16, 35) und schedule_interval den Wert 30 16 * * * hat, erfolgt die erste DAG-Ausführung um 16:30 Uhr am 6. April 2021. Da das Startdatum nach dem Zeitplanintervall am 4. April 2021 liegt, erfolgt die DAG-Ausführung nicht am 5. April 2021. Stattdessen endet das Zeitplanintervall am 5. April 2021 um 16:35 Uhr. Die nächste DAG-Ausführung wird also am folgenden Tag um 16:30 Uhr geplant.
Wenn start_date datetime(2021, 4, 4) und der schedule_interval @daily ist, wird die erste DAG-Ausführung am 5. April 2021 um 00:00 Uhr geplant.
Wenn start_date datetime(2021, 4, 4, 16, 30) und der schedule_interval 0 * * * * ist, wird die erste DAG-Ausführung am 4. April 2021 um 18:00 Uhr geplant. 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 DAG manuell auslösen, führt Airflow eine DAG-Ausführung aus. Wenn Sie beispielsweise einen DAG haben, der bereits nach einem Zeitplan ausgeführt wird, und diesen DAG manuell auslösen, führt Airflow ihn unabhängig vom tatsächlichen Zeitplan, der für den DAG angegeben wurde, einmal aus.

Console

Die DAG-UI wird in Cloud Composer 1.17.8 und höher unterstützt.

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

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

Zur Seite Umgebungen“
Wählen Sie eine Umgebung aus, um die zugehörigen Details anzusehen.
Wechseln Sie auf der Seite Umgebungsdetails zum Tab DAGs.
Klicken Sie auf den Namen eines DAG.
Klicken Sie auf der Seite DAG-Details auf DAG auslösen. Eine neue DAG-Ausführung wird erstellt.

Hinweis: Sie können keine Konfiguration für den DAG angeben, der über die Google Cloud Console ausgeführt wird.

Airflow-UI

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

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

Zur Seite Umgebungen“

Klicken Sie in der Spalte Airflow-Webserver auf den Link Airflow für Ihre Umgebung.
Melden Sie sich mit einem Google-Konto an, das über die entsprechenden Berechtigungen verfügt.
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.
Optional: Geben Sie die DAG-Ausführungskonfiguration an.
Klickbasierter Trigger

gcloud

Führen Sie in Airflow 1.10.12 oder niedriger den Befehl trigger_dag der Airflow-Befehlszeile aus:

  gcloud composer environments run ENVIRONMENT_NAME \
    --location LOCATION \
    trigger_dag -- DAG_ID

Führen Sie in Airflow 1.10.14 oder höher, einschließlich Airflow 2, den Airflow-Befehl dags trigger aus:

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

Ersetzen Sie:

ENVIRONMENT_NAME durch den Namen der Umgebung.
LOCATION durch die Region, in der sich die Umgebung befindet.
DAG_ID durch den Namen 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.