Auf Airflow-Befehlszeile zugreifen

Cloud Composer 1 Cloud Composer 2

Apache Airflow bietet eine Befehlszeile, mit der Sie Aufgaben wie das Auslösen und Verwalten von DAGs, das Abrufen von Informationen zu DAG-Ausführungen und -Aufgaben, das Hinzufügen und Löschen von Verbindungen und Nutzern ausführen können.

Über CLI-Syntaxversionen

Airflow in Cloud Composer 2 verwendet die Airflow 2-Befehlszeilensyntax.

Unterstützte Befehle der Airflow-Befehlszeile

Eine vollständige Liste der unterstützten Airflow-Befehlszeilenbefehle finden Sie in der Referenz zu gcloud composer environments run.

Hinweise

Sie müssen ausreichende Berechtigungen haben, um das gcloud-Befehlszeilentool mit Cloud Composer zu verwenden und Befehle über die Airflow-Befehlszeile auszuführen. Weitere Informationen finden Sie unter Zugriffssteuerung.
In Cloud Composer-Versionen vor 2.4.0 benötigen Sie Zugriff auf die Steuerungsebene des Clusters Ihrer Umgebung, um Befehle über die Airflow-Befehlszeile auszuführen.

Befehle der Airflow-Befehlszeile ausführen

Zum Ausführen von Befehlen der Airflow-Befehlszeile in Ihren Umgebungen verwenden Sie gcloud:

gcloud composer environments run ENVIRONMENT_NAME \
    --location LOCATION \
    SUBCOMMAND \
    -- SUBCOMMAND_ARGUMENTS

Ersetzen Sie:

ENVIRONMENT_NAME durch den Namen der Umgebung.
LOCATION durch die Region, in der sich die Umgebung befindet.
SUBCOMMAND durch einen der unterstützten Airflow-Befehlszeilenbefehle.
SUBCOMMAND_ARGUMENTS durch Argumente für den Airflow-Befehlszeilenbefehl.

Trennzeichen für Unterbefehle

Trennen Sie die Argumente für den angegebenen Airflow-Befehlszeilenbefehl durch --:

Airflow 2

Für die Syntax der Airflow-2-Befehlszeile:

Geben Sie komplexe Unterbefehle als Unterbefehl an.
Geben Sie nach dem ---Trennzeichen Argumente für zusammengesetzte Befehle als Unterbefehlsargumente an.

gcloud composer environments run example-environment \
    dags list -- --output=json

Airflow 1

Cloud Composer 2 unterstützt nur Airflow 2.

Standardort

Für die meisten gcloud composer-Befehle ist ein Speicherort erforderlich. Sie können den Speicherort mit dem Flag --location oder durch Festlegen des Standardspeicherorts angeben.

Beispiel

Beispielsweise haben Sie die Möglichkeit, mit dem folgenden Befehl einen DAG namens sample_quickstart mit der ID 5077 in Ihrer Cloud Composer-Umgebung auszulösen:

Airflow 2

gcloud composer environments run example-environment \
    --location us-central1 dags trigger -- sample_quickstart \
    --run-id=5077

Airflow 1

Cloud Composer 2 unterstützt nur Airflow 2.

Befehle in einer privaten IP-Umgebung ausführen

Ab Cloud Composer Version 2.4.0 können Sie Befehle der Airflow-Befehlszeile in einer privaten IP-Umgebung ohne zusätzliche Konfiguration ausführen. Ihr Computer benötigt zum Ausführen dieser Befehle keinen Zugriff auf den Endpunkt der Cluster-Steuerungsebene der Umgebung.

In Cloud Composer-Versionen vor 2.4.0:

Wenn Sie Befehle der Airflow-Befehlszeile in einer privaten IP-Umgebung ausführen möchten, führen Sie sie auf einem Computer aus, der auf den Endpunkt der Steuerungsebene des GKE-Cluster zugreifen kann. Die Optionen variieren dabei abhängig von Ihrer privaten Clusterkonfiguration.

Wenn der Zugriff auf öffentliche Endpunkte im Cluster Ihrer Umgebung deaktiviert ist, müssen Sie Befehle über die Airflow-Befehlszeile von einer VM im selben VPC-Netzwerk ausführen. Zur Aktivierung dieses Pfads erstellen Sie eine VM in Ihrem VPC-Netzwerk.

Wenn der Zugriff auf öffentliche Endpunkte im Cluster Ihrer Umgebung aktiviert ist, können Sie Befehle über die Airflow-Befehlszeile auch über eine Maschine mit einer öffentlichen IP-Adresse ausführen, die autorisierten Netzwerken hinzugefügt wird. Fügen Sie die externe Adresse Ihres Computers der Liste autorisierter Netzwerke Ihrer Umgebung hinzu, um den Zugriff von Ihrem Computer aus zu ermöglichen.

Befehle der Airflow-Befehlszeile über die Cloud Composer API ausführen

Ab Cloud Composer Version 2.4.0 können Sie Befehle der Airflow-Befehlszeile über die Cloud Composer API ausführen.

Befehl ausführen

Erstellen Sie eine environments.executeAirflowCommand-API-Anfrage:

{
  "environment": "projects/PROJECT_ID/locations/LOCATION/environments/ENVIRONMENT_NAME",
  "command": "AIRFLOW_COMMAND",
  "subcommand": "AIRFLOW_SUBCOMMAND",
  "parameters": [
    "SUBCOMMAND_PARAMETER"
  ]
}

Ersetzen Sie Folgendes:

PROJECT_ID: die Projekt-ID
LOCATION: Die Region, in der sich die Umgebung befindet.
ENVIRONMENT_NAME: der Name Ihrer Umgebung
AIRFLOW_COMMAND: Befehl der Airflow-Befehlszeile, den Sie ausführen möchten, z. B. dags.
AIRFLOW_SUBCOMMAND: Unterbefehl für den Airflow-Befehl, den Sie ausführen möchten, z. B. list.
(Optional) SUBCOMMAND_PARAMETER: Parameter für den Unterbefehl. Wenn Sie mehr als einen Parameter verwenden möchten, fügen Sie der Liste weitere Elemente hinzu.

Beispiel:

// POST https://composer.googleapis.com/v1/{environment=projects/*/locations/*/environments/*}:executeAirflowCommand
{
  "environment": "projects/example-project/locations/us-central1/environments/example-environment",
  "command": "dags",
  "subcommand": "list",
  "parameters": [
    "-o json",
    "--verbose"
  ]
}

Status des Abfragebefehls

Nachdem Sie einen Befehl der Airflow-Befehlszeile über die Cloud Composer API ausgeführt haben, prüfen Sie, ob der Befehl erfolgreich ausgeführt wurde. Stellen Sie dazu eine PollAirflowCommand-Anfrage und prüfen Sie die Felder in exitInfo auf Fehler und Statuscodes. Das Feld output enthält Logzeilen.

Geben Sie die von ExecuteAirflowCommandRequest zurückgegebenen Werte für executionId, pod und podNamespace an, um den Status der Befehlsausführung und Logs abzurufen:

Beispiel:

// POST https://composer.googleapis.com/v1/{environment=projects/*/locations/*/environments/*}:pollAirflowCommand
{
  "executionId": "39b82312-3a19-4d21-abac-7f8f19855ce7",
  "pod": "airflow-scheduler-1327d8cd68-hblpd",
  "podNamespace": "composer-2-4-0-airflow-2-5-3-184dadaf",
  "nextLineNumber": 1
}

Fehlerbehebung

Keine Verbindung zur Cluster-Steuerungsebene

Beim Ausführen von gcloud composer environments run- oder kubectl-Befehlen kann der folgende Fehler auftreten:

Get "https://<IP Address>/api?timeout=32s": dial tcp <IP Address>:443: i/o timeout"

Symptom: Diese Fehlermeldung zeigt an, dass keine Netzwerkverbindung zu einem Computer besteht, auf dem Sie diese Befehle ausführen.

Lösung: Folgen Sie den Richtlinien im Abschnitt Befehle in einer privaten IP-Umgebung ausführen oder folgen Sie der Anleitung im Abschnitt kubectl-Befehl überschreitet Zeitüberschreitung.