Zurückstellbare Operatoren in DAGs verwenden

Cloud Composer 3 | Cloud Composer 2 | Cloud Composer 1

Auf dieser Seite wird erläutert, wie Sie die Unterstützung für verzögerbare Operatoren in Ihrer Umgebung aktivieren und verzögerbare Operatoren in Ihren DAGs verwenden. Google Cloud

Verschiebbare Operatoren in Cloud Composer

Wenn Sie mindestens eine Triggerinstanz (oder mindestens zwei in hochverfügbaren Umgebungen) haben, können Sie zurückstellbare Operatoren und Trigger in Ihren DAGs verwenden.

Bei verschiebbaren Operatoren unterteilt Airflow die Aufgabenausführung in die folgenden Phasen:

1. Starten Sie den Vorgang. In dieser Phase belegt die Aufgabe einen Airflow-Worker-Slot. Die Aufgabe führt einen Vorgang aus, der den Job an einen anderen Dienst delegiert.

   So kann die Ausführung eines BigQuery-Jobs beispielsweise einige Sekunden bis mehrere Stunden dauern. Nach dem Erstellen des Jobs übergibt der Vorgang die Arbeits-ID (BigQuery-Job-ID) an einen Airflow-Trigger.

2. Der Trigger überwacht den Job bis zum Abschluss. In dieser Phase ist ein Worker-Slot nicht belegt. Der Airflow-Trigger hat eine asynchrone Architektur und kann Hunderte solcher Jobs verarbeiten. Wenn der Trigger erkennt, dass der Job abgeschlossen ist, sendet er ein Ereignis, das die letzte Phase auslöst.

3. In der letzten Phase führt ein Airflow-Worker einen Callback aus. Mit diesem Callback kann die Aufgabe beispielsweise als erfolgreich markiert oder ein anderer Vorgang ausgeführt und der Job so eingestellt werden, dass er wieder vom Trigger überwacht wird.

Der Auslöser ist zustandslos und daher robust gegenüber Unterbrechungen oder Neustarts. Daher sind langlaufende Jobs robust gegenüber Pod-Neustarts, es sei denn, der Neustart erfolgt während der letzten Phase, die voraussichtlich kurz sein wird.

Hinweis

• Verschiebbare Operatoren und Sensoren sind in Cloud Composer 2-Umgebungen verfügbar. Dazu sind folgende Voraussetzungen erforderlich:
  • Cloud Composer 2.0.31 und höher
  • Airflow 2.2.5, 2.3.3 und höher

Achtung: Der -Terraform-Anbieter unterstützt Airflow-Trigger derzeit nur in der Vorabversion. Verwenden Sie den Terraform-Anbieter google-beta, wenn Sie die Skalierungs- und Leistungsparameter Ihrer Umgebung ändern, auch wenn diese Änderungen nicht mit dem Auslöser zusammenhängen. Beispielsweise beim Ändern von Parametern für Airflow-Worker.

Unterstützung für verzögerbare Operatoren aktivieren

Eine Umgebungskomponente namens Airflow-Trigger überwacht asynchron alle ausgesetzten Aufgaben in Ihrer Umgebung. Nachdem ein verzögerter Vorgang einer solchen Aufgabe abgeschlossen ist, übergibt der Trigger die Aufgabe an einen Airflow-Worker.

Sie benötigen mindestens eine Triggerinstanz in Ihrer Umgebung (oder mindestens zwei in hochverfügbaren Umgebungen), um den verzögerbaren Modus in Ihren DAGs zu verwenden. Sie können die Auslöser beim Erstellen einer Umgebung konfigurieren oder die Anzahl der Auslöser und Leistungsparameter für eine vorhandene Umgebung anpassen.

Google Cloud -Operatoren, die den verzögerbaren Modus unterstützen

Nur einige Airflow-Operatoren wurden erweitert, um das verschiebbare Modell zu unterstützen. In der folgenden Liste finden Sie eine Referenz für die Operatoren im Paket airflow.providers.google.operators.cloud, die den verzögerbaren Modus unterstützen. Die Spalte mit der erforderlichen Mindestversion der airflow.providers.google.operators.cloud-Paketversion entspricht der frühesten Paketversion, bei der der jeweilige Betreiber den verzögerbaren Modus unterstützt.

Wichtig: In den folgenden Tabellen sind nicht alle Operatoren aufgeführt, die den verzögerbaren Modus unterstützen. Viele andere Betreiber von airflow.providers.google.operators.cloud und anderen Anbieterpaketen werden von Airflow und Cloud Composer unterstützt. Diese Operatoren werden auf dieser Seite nicht beschrieben. Ob ein bestimmter Anbieter den verzögerbaren Modus unterstützt, kannst du in der Dokumentation des Anbieterpakets von Airflow nachlesen.

Hinweis :BigQuery-Operatoren im verzögerbaren Modus schlagen fehl, wenn der Standort nicht auf „USA“ festgelegt ist. Das liegt an einem Fehler in einer Abhängigkeit von Airflow, nicht an BigQuery-Operatoren.

Cloud Composer-Operatoren

Name des Mobilfunkanbieters	Erforderliche apache-airflow-providers-google-Version
CloudComposerCreateEnvironmentOperator	6.4.0
CloudComposerDeleteEnvironmentOperator	6.4.0
CloudComposerUpdateEnvironmentOperator	6.4.0

BigQuery-Operatoren

Name des Mobilfunkanbieters	Erforderliche apache-airflow-providers-google-Version
BigQueryCheckOperator	8.4.0
BigQueryValueCheckOperator	8.4.0
BigQueryIntervalCheckOperator	8.4.0
BigQueryGetDataOperator	8.4.0
BigQueryInsertJobOperator	8.4.0

BigQuery Data Transfer Service-Betreiber

Name des Mobilfunkanbieters	Erforderliche apache-airflow-providers-google-Version
BigQueryDataTransferServiceStartTransferRunsOperator	8.9.0

Cloud Build-Operatoren

Name des Mobilfunkanbieters	Erforderliche apache-airflow-providers-google-Version
CloudBuildCreateBuildOperator	8.7.0

Cloud SQL-Operatoren

Name des Mobilfunkanbieters	Erforderliche apache-airflow-providers-google-Version
CloudSQLExportInstanceOperator	10.3.0

Dataflow-Operatoren

Name des Mobilfunkanbieters	Erforderliche apache-airflow-providers-google-Version
DataflowTemplatedJobStartOperator	8.9.0
DataflowStartFlexTemplateOperator	8.9.0

Cloud Data Fusion-Operatoren

Name des Mobilfunkanbieters	Erforderliche apache-airflow-providers-google-Version
CloudDataFusionStartPipelineOperator	8.9.0

Dataproc-Operatoren

Name des Mobilfunkanbieters	Erforderliche apache-airflow-providers-google-Version
DataprocCreateClusterOperator	8.9.0
DataprocDeleteClusterOperator	8.9.0
DataprocJobBaseOperator	8.4.0
DataprocInstantiateWorkflowTemplateOperator	9.0.0
DataprocInstantiateInlineWorkflowTemplateOperator	10.1.0
DataprocSubmitJobOperator	8.4.0
DataprocUpdateClusterOperator	8.9.0
DataprocCreateBatchOperator	8.9.0

Google Kubernetes Engine-Operatoren

Name des Mobilfunkanbieters	Erforderliche apache-airflow-providers-google-Version
GKEDeleteClusterOperator	9.0.0
GKECreateClusterOperator	9.0.0

AI Platform-Betreiber

Name des Mobilfunkanbieters	Erforderliche apache-airflow-providers-google-Version
MLEngineStartTrainingJobOperator	8.9.0

Zurückstellbare Operatoren in DAGs verwenden

Eine gängige Konvention für alle Google Cloud -Betreiber besteht darin, den verzögerbaren Modus mit dem booleschen Parameter deferrable zu aktivieren. Wenn ein Google Cloud-Operator diesen Parameter nicht hat, kann er nicht im verzögerbaren Modus ausgeführt werden. Bei anderen Betreibern kann eine andere Konvention gelten. Einige Betreiber von Community-Netzwerken haben beispielsweise eine separate Klasse mit dem Suffix Async im Namen.

Wichtig :Die Ordner dags/ und /plugins aus Ihrem Umgebungs-Bucket werden nicht mit dem Trigger synchronisiert. Sie können Trigger verwenden, die mit PyPI-Paketen installiert oder in einem vorinstallierten Anbieterpaket enthalten sind.

Im folgenden Beispiel-DAG wird der Operator DataprocSubmitJobOperator im verzögerbaren Modus verwendet:

PYSPARK_JOB = {
    "reference": { "project_id": "PROJECT_ID" },
    "placement": { "cluster_name": "PYSPARK_CLUSTER_NAME" },
    "pyspark_job": {
        "main_python_file_uri": "gs://dataproc-examples/pyspark/hello-world/hello-world.py"
    },
}

DataprocSubmitJobOperator(
        task_id="dataproc-deferrable-example",
        job=PYSPARK_JOB,
        deferrable=True,
    )

Trigger-Logs ansehen

Der Trigger generiert Protokolle, die zusammen mit Protokollen anderer Umgebungskomponenten verfügbar sind. Weitere Informationen zum Aufrufen von Umgebungslogs finden Sie unter Logs ansehen.

Monitor-Trigger

Weitere Informationen zum Überwachen der Triggerkomponente finden Sie unter Airflow-Messwerte.

Sie können nicht nur den Auslöser überwachen, sondern auch die Anzahl der verschobenen Aufgaben im Messwert Unerledigte Aufgabe im Monitoring-Dashboard Ihrer Umgebung prüfen.

Nächste Schritte

• Fehlerbehebung bei Airflow-Triggern
• Messwerte für Airflow-Trigger
• Airflow-Trigger-Logs