Cloud Composer 1 | Cloud Composer 2 | Cloud Composer 3
In diesem Leitfaden erfahren Sie, wie Sie einen gerichteten azyklischen Graphen in Apache Airflow schreiben (DAG), der in einer Cloud Composer-Umgebung ausgeführt wird.
Da Apache Airflow keine starke DAG- und Aufgabenisolierung bietet, empfehlen wir, separate Produktions- und Testumgebungen um DAG-Interferenzen zu vermeiden. Weitere Informationen finden Sie unter DAGs testen.
Airflow-DAG strukturieren
Ein Airflow-DAG wird in einer Python-Datei definiert und besteht aus folgenden Komponenten: Komponenten:
- DAG-Definition
- Airflow-Operatoren
- Beziehungen zu Betreibern
Die folgenden Code-Snippets zeigen Beispiele für die einzelnen Komponenten außerhalb des Kontexts.
Eine DAG-Definition
Im folgenden Beispiel wird ein Airflow-DAG veranschaulicht: Definition:
Airflow 2
Airflow 1
Operatoren und Aufgaben
Airflow Operator beschreiben die auszuführende Arbeit. Eine Aufgabe task ist eine spezifische Instanz eines Operators.
Airflow 2
Airflow 1
Aufgabenbeziehungen
Aufgabenbeziehungen beschreiben die Reihenfolge, in der die Arbeit ausgeführt werden muss.
Airflow 2
Airflow 1
Beispiel für einen vollständigen DAG-Workflow in Python
Der folgende Workflow ist eine vollständige, funktionierende DAG-Vorlage, die aus
zwei Aufgaben: die Aufgabe hello_python
und die Aufgabe goodbye_bash
:
Airflow 2
Airflow 1
Weitere Informationen zum Definieren von Airflow-DAGs finden Sie in der Airflow-Anleitung und Airflow-Konzepte
Airflow-Operatoren
Die folgenden Beispiele enthalten einige beliebte Airflow-Operatoren. Für eine Referenz zu Airflow-Operatoren finden Sie in der Operator- und Hook-Referenz und Anbieterindex.
BashOperator
Mit dem BashOperator können Sie Befehlszeilenprogramme ausführen.
Airflow 2
Airflow 1
Cloud Composer führt die bereitgestellten Befehle in einem Bash-Skript auf einem Airflow-Worker Der Worker ist ein Docker-Container auf Basis von Debian und enthält mehrere Pakete.
PythonOperator
Verwenden Sie die Methode PythonOperator verwendet, um beliebigen Python-Code auszuführen.
Cloud Composer führt den Python-Code in einem Container aus, der Pakete für die in Ihrer Umgebung verwendete Cloud Composer-Image-Version enthält.
Informationen zum Installieren weiterer Python-Pakete finden Sie unter Python-Abhängigkeiten installieren.
Google Cloud-Operatoren
Verwenden Sie zum Ausführen von Aufgaben, die Google Cloud-Produkte verwenden, den Google Cloud-Airflow-Operatoren. Beispiel: BigQuery-Operatoren Daten in BigQuery abfragen und verarbeiten können.
Es gibt viele weitere Airflow-Operatoren für Google Cloud und einzelne Dienste von Google Cloud. Weitere Informationen finden Sie unter Eine vollständige Liste der Google Cloud-Operatoren finden Sie hier.
Airflow 2
Airflow 1
EmailOperator
Verwenden Sie die Methode EmailOperator verwendet, um E-Mails von einem DAG zu senden. Senden E-Mails aus einer Cloud Composer-Umgebung zu senden, Konfigurieren Sie Ihre Umgebung für die Verwendung von SendGrid.
Airflow 2
Airflow 1
Benachrichtigungen bei Ausfall des Betreibers
Zum Senden einer E-Mail-Benachrichtigung, wenn ein Operator im DAG fehlerhaft ist, legen Sie für email_on_failure
den Wert True
fest. Zum Senden von E-Mails aus einer Cloud Composer-Umgebung müssen Sie Ihre Umgebung für die Verwendung von SendGrid konfigurieren.
Airflow 2
Airflow 1
Richtlinien für DAG-Workflows
Platzieren Sie benutzerdefinierte Python-Bibliotheken im ZIP-Archiv eines DAG in einem verschachtelten Verzeichnis. Platzieren Sie Bibliotheken nicht auf der obersten Ebene des DAG-Verzeichnisses.
Wenn Airflow den Ordner
dags/
scannt, sucht Airflow nur nach DAGs in Python-Module, die sich auf der obersten Ebene des DAGs-Ordners und in der obersten Ebene des Ebene eines ZIP-Archivs, das sich auch im obersten Ordnerdags/
befindet. Wenn Airflow findet in einem ZIP-Archiv ein Python-Modul, das keine sowohl denairflow
- als auch denDAG
-Teilstrings verwenden, beendet Airflow die Verarbeitung der ZIP-Datei -Archiv. Airflow gibt nur die DAGs zurück, die bis zu diesem Zeitpunkt gefunden wurden.Verwenden Sie Airflow 2 anstelle von Airflow 1.
Die Airflow-Community veröffentlicht keine neuen Neben- oder Patch-Releases mehr für Airflow 1.
Achten Sie aus Gründen der Fehlertoleranz darauf, nicht mehrere DAG-Objekte im gleichen Python-Modul zu definieren.
Verwenden Sie SubDAGs nicht. Stattdessen Aufgaben in DAGs gruppieren.
Platzieren Sie Dateien, die zum Zeitpunkt der DAG-Analyse erforderlich sind, im Ordner
dags/
, nicht im Ordnerdata/
.Testen Sie entwickelte oder geänderte DAGs, wie in den Anleitung zum Testen von DAGs.
Achten Sie darauf, dass sich entwickelte DAGs nicht erhöhen DAG-Analysezeiten zu oft.
Airflow-Aufgaben können aus verschiedenen Gründen fehlschlagen. Um Ausfälle der ganze DAG-Ausführungen ausführen möchten, sollten Sie Wiederholungsversuche für Aufgaben aktivieren. Wenn Sie die maximale Anzahl von Wiederholungen auf
0
festlegen, werden keine Wiederholungen ausgeführt.Wir empfehlen, die Option
default_task_retries
mit einem anderen Wert für die Wiederholungen von Aufgaben als0
zu überschreiben. Darüber hinaus können Sie die Parameterretries
auf Aufgabenebene.Wenn Sie GPUs in Ihren Airflow-Aufgaben verwenden möchten, erstellen Sie einen separaten GKE-Cluster, der auf Knoten mit Maschinen mit GPUs basiert. Verwenden Sie GKEStartPodOperator, um Ihre Aufgaben auszuführen.
Führen Sie keine CPU- und arbeitsspeicherintensiven Aufgaben im Knotenpool des Clusters aus, andere Airflow-Komponenten (Planer, Worker, Webserver) ausgeführt werden. Verwenden Sie stattdessen KubernetesPodOperator oder GKEStartPodOperator.
Wenn Sie DAGs in einer Umgebung bereitstellen, laden Sie nur die Dateien hoch, für die Interpretation und Ausführung von DAGs in den Ordner
/dags
verschieben.Begrenzen Sie die Anzahl der DAG-Dateien im Ordner „
/dags
“.Airflow parst kontinuierlich DAGs im Ordner
/dags
. Das Parsing ist ein Prozess, der den DAGs-Ordner und die Anzahl der Dateien durchsucht, (mit ihren Abhängigkeiten) die Leistung beeinträchtigt. von DAG-Parsing und Aufgabenplanung. Es ist viel effizienter, 100 Dateien mit jeweils 100 DAGs über 10.000 Dateien mit jeweils einem DAG. wird eine Optimierung empfohlen. Diese Optimierung ist ein Gleichgewicht zwischen Parsing-Zeit und Effizienz von DAG-Erstellung und -VerwaltungSie können auch 10.000 DAG-Dateien bereitstellen, 100 ZIP-Dateien mit jeweils 100 DAG-Dateien erstellen
Wenn Sie mehr als 10.000 DAG-Dateien haben, könnte eine gute Option sein, DAGs programmatisch zu generieren. Beispiel: können Sie eine einzelne Python-DAG-Datei implementieren, DAG-Objekte (z. B. 20 oder 100 DAG-Objekte)
Verworfene Airflow-Operatoren vermeiden
Die in der folgenden Tabelle aufgeführten Operatoren wurden eingestellt. Einige dieser Operatoren wurden in frühen Versionen von Cloud Composer 1 unterstützt. Vermeiden Sie die Verwendung in DAGs zu koordinieren. Verwende stattdessen aktuelle Alternativen.
Veralteter Operator | Zu verwendender Operator |
---|---|
BigQueryExecuteQueryOperator | BigQueryInsertJobOperator |
BigQueryPatchDatasetOperator | BigQueryUpdateTableOperator |
DataflowCreateJavaJobOperator | BeamRunJavaPipelineOperator |
DataflowCreatePythonJobOperator | BeamRunPythonPipelineOperator |
DataprocScaleClusterOperator | DataprocUpdateClusterOperator |
DataprocSubmitPigJobOperator | DataprocSubmitJobOperator |
DataprocSubmitSparkSqlJobOperator | DataprocSubmitJobOperator |
DataprocSubmitSparkJobOperator | DataprocSubmitJobOperator |
DataprocSubmitHadoopJobOperator | DataprocSubmitJobOperator |
DataprocSubmitPySparkJobOperator | DataprocSubmitJobOperator |
MLEngineManageModelOperator | MLEngineCreateModelOperator, MLEngineGetModelOperator |
MLEngineManageVersionOperator | MLEngineCreateVersion, MLEngineSetDefaultVersion, MLEngineListVersions, MLEngineDeleteVersion |
GCSObjectsWtihPrefixExistenceSensor | GCSObjectsWithPrefixExistenceSensor |
FAQs zum Schreiben von DAGs
Wie minimiere ich Codewiederholungen, wenn ich die gleichen oder ähnliche Aufgaben in mehreren DAGs ausführen möchte?
Wir empfehlen, Bibliotheken und Wrapper für Codewiederholungen zu minimieren.
Wie kann ich Code in mehreren DAG-Dateien wiederverwenden?
Platzieren Sie Ihre Dienstprogrammfunktionen in einem
lokale Python-Bibliothek
und importieren die Funktionen. Sie können in allen DAGs, die sich im dags/
-Ordner Ihres Buckets befinden, auf die Funktionen verweisen.
Wie minimiere ich das Risiko unterschiedlicher Definitionen?
Angenommen, es gibt zwei Teams, die Rohdaten zu Umsatzkennzahlen zusammenfassen möchten. Die Teams schreiben zwei geringfügig unterschiedliche Aufgaben für den gleichen Sachverhalt. Definieren Sie Bibliotheken für die Arbeit mit den Umsatzdaten, sodass diejenigen, die DAGs implementieren, die Definition des zusammengefassten Umsatzes eindeutig festlegen müssen.
Wie lege ich Abhängigkeiten zwischen DAGs fest?
Das hängt davon ab, wie Sie die Abhängigkeit definieren möchten.
Wenn Sie zwei DAGs haben (DAG A und DAG B) und DAG B nach dem DAG ausgelöst werden soll
A: Sie können eine
TriggerDagRunOperator
am Ende von DAG A.
Wenn DAG B nur von einem von DAG A generierten Artefakt abhängt (z. B. eine Pub/Sub-Meldung), ist ein Sensor möglicherweise besser geeignet.
Wenn DAG B eng mit DAG A integriert ist, können Sie die beiden DAGs möglicherweise in einen DAG zusammenführen.
Wie übergebe ich eindeutige Ausführungs-IDs an einen DAG und die zugehörigen Aufgaben?
Angenommen, es sollen Dataproc-Clusternamen und -Dateipfade übergeben werden.
In diesem Fall können Sie eine zufällige eindeutige ID generieren und dafür str(uuid.uuid4())
in einem PythonOperator
zurückgeben. Dadurch wird die ID
XComs
, damit Sie in anderen Operatoren auf die ID verweisen können
mithilfe von Vorlagenfeldern.
Prüfen Sie vor dem Generieren einer uuid
, ob eine DagRun-spezifische ID sinnvoller wäre. Sie können auf diese IDs in Jinja-Substitutionen auch mit Makros verweisen.
Wie trenne ich Aufgaben in einem DAG?
Eine Aufgabe sollte eine idempotente Arbeitseinheit sein. Vermeiden Sie es deshalb, einen aus mehreren Schritten bestehenden Workflow in eine einzelne Aufgabe aufzunehmen, z. B. in ein komplexes Programm, das in einem PythonOperator
ausgeführt wird.
Soll ich mehrere Aufgaben in einem einzelnen DAG definieren, um Daten aus mehreren Quellen zusammenzufassen?
Angenommen, ich habe mehrere Tabellen mit Rohdaten und möchte tägliche Zusammenfassungen für jede einzelne Tabelle erstellen. Die Aufgaben sind nicht voneinander abhängig. Soll ich eine Aufgabe und einen DAG für jede Tabelle oder einen allgemeinen DAG erstellen?
Wenn es für Sie kein Problem ist, dass jede Aufgabe die gleichen Attribute auf DAG-Ebene verwendet (z. B. schedule_interval
), ist es sinnvoll, mehrere Aufgaben in einem einzigen DAG zu definieren. Andernfalls können zur Minimierung der Codewiederholung mehrere DAGs aus einem einzigen Python-Modul generiert werden. Dazu platzieren Sie diese in den globalen globals()
des Moduls.
Wie beschränke ich die Anzahl gleichzeitiger Aufgaben, die in einem DAG ausgeführt werden?
Ich möchte z. B. vermeiden, dass API-Nutzungslimits und -kontingente überschritten oder zu viele Prozesse gleichzeitig ausgeführt werden.
Sie können Airflow-Pools in der Airflow-Web-UI und verknüpfte Aufgaben mit vorhandenen Pools in Ihren DAGs.
FAQs zur Verwendung von Operatoren
Soll ich den DockerOperator
verwenden?
Wir raten davon ab,
Das DockerOperator
, es sei denn, es wird zum Starten von
Container in einer Docker-Remote-Installation (nicht im
Cluster). In einer Cloud Composer-Umgebung hat der Operator keinen Zugriff auf Docker-Daemons.
Verwenden Sie stattdessen KubernetesPodOperator
oder
GKEStartPodOperator
Diese Operatoren starten Kubernetes-Pods in
Kubernetes- bzw. GKE-Cluster. Beachten Sie, dass wir
empfehlen, Pods im Cluster einer Umgebung zu starten, da dies
und Ressourcenwettbewerbe bieten.
Soll ich den SubDagOperator
verwenden?
Die Verwendung von SubDagOperator
wird nicht empfohlen.
Verwenden Sie Alternativen, wie unter Aufgaben gruppieren beschrieben.
Soll ich Python-Code nur in PythonOperators
ausführen, um Python-Operatoren vollständig zu trennen?
Abhängig von Ihrem Ziel haben Sie mehrere Optionen.
Falls Ihr einziges Ziel ist, separate Python-Abhängigkeiten beizubehalten, können Sie PythonVirtualenvOperator
verwenden.
Erwägen Sie die Verwendung der KubernetesPodOperator
. Mit diesem Operator können Sie
um Kubernetes-Pods zu definieren und die Pods in anderen Clustern auszuführen.
Wie füge ich benutzerdefinierte binäre oder Nicht-PyPI-Pakete hinzu?
Sie können dazu Pakete installieren, die in privaten Paket-Repositories gehostet werden,
Wie übergebe ich Argumente einheitlich an einen DAG und die zugehörigen Aufgaben?
Sie können die integrierte Unterstützung von Airflow für Jinja-Vorlagen zur Übergabe verwendbarer Argumente in Vorlagenfeldern.
Wann findet die Vorlagenersetzung statt?
Die Vorlagen werden auf den Airflow-Workern unmittelbar vor dem Aufruf der pre_execute
-Funktion eines Operators ersetzt. In der Praxis bedeutet das, dass Vorlagen
erst kurz vor der Ausführung einer Aufgabe ersetzt.
Wie kann ich erkennen, welche Operatorargumente die Vorlagenersetzung unterstützen?
Operatorargumente, die die Jinja2-Vorlagenersetzung unterstützen, sind explizit gekennzeichnet sind.
Suchen Sie in der Operatordefinition nach dem Feld template_fields
.
Sie enthält eine Liste von Argumentnamen, für die die Vorlagenersetzung durchgeführt wird.
Siehe zum Beispiel
den BashOperator
, der die Vorlagenerstellung für
die Argumente bash_command
und env
.
Nächste Schritte
- Fehlerbehebung bei DAGs
- Fehlerbehebung beim Scheduler
- Google-Betreiber
- Google Cloud-Operatoren
- Apache Airflow-Anleitung