Rücksetzbare Operatoren in DAGs verwenden

Cloud Composer 1 Cloud Composer 2

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

Rücksetzbare Operatoren in Cloud Composer

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

Bei zurückstellbaren Operatoren teilt Airflow die Aufgabenausführung in die folgenden Phasen auf:

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

   Die Ausführung eines BigQuery-Jobs kann beispielsweise von einigen Sekunden bis zu mehreren Stunden dauern. Nach dem Erstellen des Jobs übergibt der Vorgang die Arbeitskennung (BigQuery-Job-ID) an einen Airflow-Trigger.

2. Der Trigger überwacht den Job, bis er abgeschlossen ist. 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. Durch diesen Callback kann beispielsweise die Aufgabe als erfolgreich markiert oder ein weiterer Vorgang ausgeführt und der Job noch einmal vom Trigger überwacht werden.

Der Trigger ist zustandslos und daher robust gegenüber Unterbrechungen oder Neustarts. Aus diesem Grund sind Jobs mit langer Ausführungszeit gegenüber Pod-Neustarts resistent, es sei denn, der Neustart erfolgt in der letzten Phase, die voraussichtlich kurz ist.

Hinweise

• Rücksetzbare Operatoren und Sensoren sind in Cloud Composer 2-Umgebungen verfügbar und erfordern Folgendes:
  • Cloud Composer 2.0.31 und höher
  • Airflow 2.2.5, 2.3.3 und höher

Achtung: Die Unterstützung des Terraform-Anbieters für Airflow-Trigger befindet sich in der Vorabversion. Verwenden Sie den Terraform-Anbieter google-beta, wenn Sie Skalierungs- und Leistungsparameter Ihrer Umgebung ändern, auch wenn diese Änderungen nicht mit dem Trigger zusammenhängen. beispielsweise beim Ändern von Parametern für Airflow-Worker.

Unterstützung für zurückstellbare Operatoren aktivieren

Eine Umgebungskomponente namens Airflow-Trigger überwacht alle ausgesetzten Aufgaben in Ihrer Umgebung asynchron. Nachdem ein zurückgestellter Vorgang aus einer solchen Aufgabe abgeschlossen wurde, übergibt der Trigger die Aufgabe an einen Airflow-Worker.

Sie benötigen mindestens eine Triggerinstanz in Ihrer Umgebung (oder mindestens zwei in einer äußerst robusten Umgebung), um den zurückstellbaren Modus in Ihren DAGs zu verwenden. Sie können die Trigger beim Erstellen einer Umgebung konfigurieren oder die Anzahl der Trigger und Leistungsparameter für eine vorhandene Umgebung anpassen.

Google Cloud-Operatoren, die den Modus „Verzögerbar“ unterstützen

Es wurden nur einige Airflow-Operatoren zur Unterstützung des zurückgestellten Modells erweitert. Die folgende Liste ist eine Referenz für die Operatoren im Paket airflow.providers.google.operators.cloud, die den Modus „deferrable“ unterstützen. Die Spalte mit der mindestens erforderlichen airflow.providers.google.operators.cloud-Paketversion gibt die früheste Paketversion an, in der dieser Operator den Modus zum Aussetzen unterstützt.

Wichtig :In den folgenden Tabellen sind nicht alle Operatoren aufgeführt, die den Modus „Aussetzender“ unterstützen. Viele andere Operatoren 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. Informationen dazu, ob ein bestimmter Operator den Modus „Verbindbar“ unterstützt, finden Sie in der Dokumentation zum Anbieterpaket von Airflow.

Hinweis: BigQuery-Operatoren im abweisbaren Modus schlagen fehl, wenn der Standort nicht auf US festgelegt ist. Dies geschieht aufgrund eines Fehlers in einer Abhängigkeit von Airflow und nicht in BigQuery-Operatoren.

Cloud Composer-Operatoren

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

BigQuery-Operatoren

Name des Betreibers	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-Operatoren

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

Cloud Build-Operatoren

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

Cloud SQL-Operatoren

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

Dataflow-Operatoren

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

Cloud Data Fusion-Operatoren

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

Dataproc-Operatoren

Name des Betreibers	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 Betreibers	Erforderliche apache-airflow-providers-google-Version
GKEDeleteClusterOperator	9.0.0
GKECreateClusterOperator	9.0.0

AI Platform-Operatoren

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

Rücksetzbare Operatoren in DAGs verwenden

Eine gängige Konvention für alle Google Cloud-Operatoren besteht darin, dass der zurückstellbare Modus mit dem booleschen Parameter deferrable aktiviert wird. Wenn ein Google Cloud-Operator diesen Parameter nicht hat, kann er nicht im Modus „Aussetzen“ ausgeführt werden. Andere Operatoren können eine andere Konvention haben. Einige Community-Operatoren 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 Modus zum Aussetzen 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,
    )

Triggererlogs ansehen

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

Monitor-Trigger

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

Zusätzlich zum Monitoring des Triggers können Sie die Anzahl der ausgesetzten Aufgaben in den Messwerten für unbeendete Aufgaben im Monitoring-Dashboard Ihrer Umgebung prüfen.

Nächste Schritte

• Airflow-Triggerprobleme beheben
• Airflow-Triggerer-Messwerte
• Airflow-Triggerlogs