Airflow-DAGs schreiben

Cloud Composer 1 Cloud Composer 2

In dieser Anleitung wird beschrieben, wie Sie einen gerichteten azyklischen Graphen (Directed Acyclic Graph, DAG) in Apache Airflow für die Ausführung in einer Cloud Composer-Umgebung schreiben.

Da Apache Airflow keine starke DAG- und Aufgabenisolation bietet, empfehlen wir die Verwendung separater 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 den folgenden Komponenten:

  • DAG-Definition
  • Airflow-Operatoren
  • Operatorbeziehungen

Die folgenden Code-Snippets zeigen Beispiele für jede Komponente außerhalb des Kontexts.

Eine DAG-Definition

Im folgenden Beispiel wird eine DAG-Definition veranschaulicht:

Airflow 2 

import datetime

from airflow import models

default_dag_args = {
    # The start_date describes when a DAG is valid / can be run. Set this to a
    # fixed point in time rather than dynamically, since it is evaluated every
    # time a DAG is parsed. See:
    # https://airflow.apache.org/faq.html#what-s-the-deal-with-start-date
    "start_date": datetime.datetime(2018, 1, 1),
}

# Define a DAG (directed acyclic graph) of tasks.
# Any task you create within the context manager is automatically added to the
# DAG object.
with models.DAG(
    "composer_sample_simple_greeting",
    schedule_interval=datetime.timedelta(days=1),
    default_args=default_dag_args,
) as dag:

Airflow 1 

import datetime

from airflow import models

default_dag_args = {
    # The start_date describes when a DAG is valid / can be run. Set this to a
    # fixed point in time rather than dynamically, since it is evaluated every
    # time a DAG is parsed. See:
    # https://airflow.apache.org/faq.html#what-s-the-deal-with-start-date
    "start_date": datetime.datetime(2018, 1, 1),
}

# Define a DAG (directed acyclic graph) of tasks.
# Any task you create within the context manager is automatically added to the
# DAG object.
with models.DAG(
    "composer_sample_simple_greeting",
    schedule_interval=datetime.timedelta(days=1),
    default_args=default_dag_args,
) as dag:

Operatoren und Aufgaben

Operatoren beschreiben die auszuführende Arbeit. Eine Aufgabe ist eine spezifische Instanz eines Operators.

Airflow 2 

from airflow.operators import bash_operator
from airflow.operators import python_operator

    def greeting():
        import logging

        logging.info("Hello World!")

    # An instance of an operator is called a task. In this case, the
    # hello_python task calls the "greeting" Python function.
    hello_python = python_operator.PythonOperator(
        task_id="hello", python_callable=greeting
    )

    # Likewise, the goodbye_bash task calls a Bash script.
    goodbye_bash = bash_operator.BashOperator(
        task_id="bye", bash_command="echo Goodbye."
    )

Airflow 1 

from airflow.operators import bash_operator
from airflow.operators import python_operator

    def greeting():
        import logging

        logging.info("Hello World!")

    # An instance of an operator is called a task. In this case, the
    # hello_python task calls the "greeting" Python function.
    hello_python = python_operator.PythonOperator(
        task_id="hello", python_callable=greeting
    )

    # Likewise, the goodbye_bash task calls a Bash script.
    goodbye_bash = bash_operator.BashOperator(
        task_id="bye", bash_command="echo Goodbye."
    )

Aufgabenbeziehungen

Aufgabenbeziehungen beschreiben die Reihenfolge, in der die Arbeit ausgeführt werden muss.

Airflow 2 

# Define the order in which the tasks complete by using the >> and <<
# operators. In this example, hello_python executes before goodbye_bash.
hello_python >> goodbye_bash

Airflow 1 

# Define the order in which the tasks complete by using the >> and <<
# operators. In this example, hello_python executes before goodbye_bash.
hello_python >> goodbye_bash

Vollständiges Beispiel für DAG-Workflow in Python

Der folgende Workflow ist eine vollständige funktionierende DAG-Vorlage, die aus zwei Aufgaben besteht: einer hello_python-Aufgabe und einer goodbye_bash-Aufgabe:

Airflow 2 


import datetime

from airflow import models

from airflow.operators import bash_operator
from airflow.operators import python_operator

default_dag_args = {
    # The start_date describes when a DAG is valid / can be run. Set this to a
    # fixed point in time rather than dynamically, since it is evaluated every
    # time a DAG is parsed. See:
    # https://airflow.apache.org/faq.html#what-s-the-deal-with-start-date
    "start_date": datetime.datetime(2018, 1, 1),
}

# Define a DAG (directed acyclic graph) of tasks.
# Any task you create within the context manager is automatically added to the
# DAG object.
with models.DAG(
    "composer_sample_simple_greeting",
    schedule_interval=datetime.timedelta(days=1),
    default_args=default_dag_args,
) as dag:
    def greeting():
        import logging

        logging.info("Hello World!")

    # An instance of an operator is called a task. In this case, the
    # hello_python task calls the "greeting" Python function.
    hello_python = python_operator.PythonOperator(
        task_id="hello", python_callable=greeting
    )

    # Likewise, the goodbye_bash task calls a Bash script.
    goodbye_bash = bash_operator.BashOperator(
        task_id="bye", bash_command="echo Goodbye."
    )

    # Define the order in which the tasks complete by using the >> and <<
    # operators. In this example, hello_python executes before goodbye_bash.
    hello_python >> goodbye_bash

Airflow 1 


import datetime

from airflow import models

from airflow.operators import bash_operator
from airflow.operators import python_operator

default_dag_args = {
    # The start_date describes when a DAG is valid / can be run. Set this to a
    # fixed point in time rather than dynamically, since it is evaluated every
    # time a DAG is parsed. See:
    # https://airflow.apache.org/faq.html#what-s-the-deal-with-start-date
    "start_date": datetime.datetime(2018, 1, 1),
}

# Define a DAG (directed acyclic graph) of tasks.
# Any task you create within the context manager is automatically added to the
# DAG object.
with models.DAG(
    "composer_sample_simple_greeting",
    schedule_interval=datetime.timedelta(days=1),
    default_args=default_dag_args,
) as dag:
    def greeting():
        import logging

        logging.info("Hello World!")

    # An instance of an operator is called a task. In this case, the
    # hello_python task calls the "greeting" Python function.
    hello_python = python_operator.PythonOperator(
        task_id="hello", python_callable=greeting
    )

    # Likewise, the goodbye_bash task calls a Bash script.
    goodbye_bash = bash_operator.BashOperator(
        task_id="bye", bash_command="echo Goodbye."
    )

    # Define the order in which the tasks complete by using the >> and <<
    # operators. In this example, hello_python executes before goodbye_bash.
    hello_python >> goodbye_bash

Weitere Informationen zum Definieren von Airflow-DAGs finden Sie in der Airflow-Anleitung und in den Airflow-Konzepten.

Airflow-Operatoren

Die folgenden Beispiele enthalten einige beliebte Airflow-Operatoren. Eine verbindliche Referenz zu Airflow-Operatoren finden Sie in der Referenz zu Operatoren und Hooks und im Index „Providers“.

BashOperator

Mit dem BashOperator können Sie Befehlszeilenprogramme ausführen.

Airflow 2 

from airflow.operators import bash

    # Create BigQuery output dataset.
    make_bq_dataset = bash.BashOperator(
        task_id="make_bq_dataset",
        # Executing 'bq' command requires Google Cloud SDK which comes
        # preinstalled in Cloud Composer.
        bash_command=f"bq ls {bq_dataset_name} || bq mk {bq_dataset_name}",
    )

Airflow 1 

from airflow.operators import bash_operator

    # Create BigQuery output dataset.
    make_bq_dataset = bash_operator.BashOperator(
        task_id="make_bq_dataset",
        # Executing 'bq' command requires Google Cloud SDK which comes
        # preinstalled in Cloud Composer.
        bash_command=f"bq ls {bq_dataset_name} || bq mk {bq_dataset_name}",
    )

Cloud Composer führt die bereitgestellten Befehle in einem Bash-Skript auf einem Airflow-Worker aus. Der Worker ist ein Debian-basierter Docker-Container und enthält mehrere Pakete.

PythonOperator

Verwenden Sie den PythonOperator, 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 die Google Cloud-Airflow-Operatoren, um Aufgaben auszuführen, die Google Cloud-Produkte verwenden. BigQuery-Operatoren fragen beispielsweise Daten in BigQuery ab und verarbeiten sie.

Es gibt viele weitere Airflow-Operatoren für Google Cloud und einzelne Dienste von Google Cloud. Die vollständige Liste finden Sie unter Google Cloud-Operatoren.

Airflow 2 

from airflow.providers.google.cloud.operators import bigquery
from airflow.providers.google.cloud.transfers import bigquery_to_gcs

    bq_recent_questions_query = bigquery.BigQueryInsertJobOperator(
        task_id="bq_recent_questions_query",
        configuration={
            "query": {
                "query": RECENT_QUESTIONS_QUERY,
                "useLegacySql": False,
                "destinationTable": {
                    "projectId": project_id,
                    "datasetId": bq_dataset_name,
                    "tableId": bq_recent_questions_table_id,
                },
            }
        },
        location=location,
    )

Airflow 1 

from airflow.contrib.operators import bigquery_operator

    # Query recent StackOverflow questions.
    bq_recent_questions_query = bigquery_operator.BigQueryOperator(
        task_id="bq_recent_questions_query",
        sql="""
        SELECT owner_display_name, title, view_count
        FROM `bigquery-public-data.stackoverflow.posts_questions`
        WHERE creation_date < CAST('{max_date}' AS TIMESTAMP)
            AND creation_date >= CAST('{min_date}' AS TIMESTAMP)
        ORDER BY view_count DESC
        LIMIT 100
        """.format(
            max_date=max_query_date, min_date=min_query_date
        ),
        use_legacy_sql=False,
        destination_dataset_table=bq_recent_questions_table_id,
    )

EmailOperator

Verwenden Sie den EmailOperator, um E-Mails von einem DAG zu senden. Zum Senden von E-Mails aus einer Cloud Composer-Umgebung müssen Sie die Umgebung für die Verwendung von SendGrid konfigurieren.

Airflow 2 

from airflow.operators import email

    # Send email confirmation (you will need to set up the email operator
    # See https://cloud.google.com/composer/docs/how-to/managing/creating#notification
    # for more info on configuring the email operator in Cloud Composer)
    email_summary = email.EmailOperator(
        task_id="email_summary",
        to="{{var.value.email}}",
        subject="Sample BigQuery notify data ready",
        html_content="""
        Analyzed Stack Overflow posts data from {min_date} 12AM to {max_date}
        12AM. The most popular question was '{question_title}' with
        {view_count} views. Top 100 questions asked are now available at:
        {export_location}.
        """.format(
            min_date=min_query_date,
            max_date=max_query_date,
            question_title=(
                "{{ ti.xcom_pull(task_ids='bq_read_most_popular', "
                "key='return_value')[0][0] }}"
            ),
            view_count=(
                "{{ ti.xcom_pull(task_ids='bq_read_most_popular', "
                "key='return_value')[0][1] }}"
            ),
            export_location=output_file,
        ),
    )

Airflow 1 

from airflow.operators import email_operator

    # Send email confirmation
    email_summary = email_operator.EmailOperator(
        task_id="email_summary",
        to="{{var.value.email}}",
        subject="Sample BigQuery notify data ready",
        html_content="""
        Analyzed Stack Overflow posts data from {min_date} 12AM to {max_date}
        12AM. The most popular question was '{question_title}' with
        {view_count} views. Top 100 questions asked are now available at:
        {export_location}.
        """.format(
            min_date=min_query_date,
            max_date=max_query_date,
            question_title=(
                "{{ ti.xcom_pull(task_ids='bq_read_most_popular', "
                "key='return_value')[0][0] }}"
            ),
            view_count=(
                "{{ ti.xcom_pull(task_ids='bq_read_most_popular', "
                "key='return_value')[0][1] }}"
            ),
            export_location=output_file,
        ),
    )

Benachrichtigungen bei Ausfall des Bedieners

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 

from airflow import models

default_dag_args = {
    "start_date": yesterday,
    # Email whenever an Operator in the DAG fails.
    "email": "{{var.value.email}}",
    "email_on_failure": True,
    "email_on_retry": False,
    "retries": 1,
    "retry_delay": datetime.timedelta(minutes=5),
    "project_id": project_id,
}

with models.DAG(
    "composer_sample_bq_notify",
    schedule_interval=datetime.timedelta(weeks=4),
    default_args=default_dag_args,
) as dag:

Airflow 1 

from airflow import models

default_dag_args = {
    "start_date": yesterday,
    # Email whenever an Operator in the DAG fails.
    "email": "{{var.value.email}}",
    "email_on_failure": True,
    "email_on_retry": False,
    "retries": 1,
    "retry_delay": datetime.timedelta(minutes=5),
    "project_id": "{{var.value.gcp_project}}",
}

with models.DAG(
    "composer_sample_bq_notify",
    schedule_interval=datetime.timedelta(weeks=4),
    default_args=default_dag_args,
) as dag:

Richtlinien für DAG-Workflows

  1. 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 in Python-Modulen nach DAGs, die sich auf der obersten Ebene des DAGs-Ordners und auf der obersten Ebene eines ZIP-Archivs befinden, das sich ebenfalls im übergeordneten Ordner dags/ befindet. Wenn Airflow in einem ZIP-Archiv auf ein Python-Modul stößt, das weder airflow- noch DAG-Teilstrings enthält, beendet Airflow die Verarbeitung des ZIP-Archivs. Airflow gibt nur die DAGs zurück, die bis zu diesem Zeitpunkt gefunden wurden.

  2. Verwenden Sie Airflow 2 anstelle von Airflow 1.

    Die Airflow-Community veröffentlicht keine neuen Neben- oder Patchversionen für Airflow 1 mehr.

  3. Achten Sie aus Gründen der Fehlertoleranz darauf, nicht mehrere DAG-Objekte im gleichen Python-Modul zu definieren.

  4. Verwenden Sie keine SubDAGs. Gruppieren Sie stattdessen Aufgaben in DAGs.

  5. Legen Sie Dateien, die zum Zeitpunkt der DAG-Analyse erforderlich sind, im Ordner dags/ und nicht im Ordner data/ ab.

  6. Einheitentests für Ihre DAGs implementieren

  7. Testen Sie entwickelte oder geänderte DAGs, wie in der Anleitung zum Testen von DAGs empfohlen.

  8. Achten Sie darauf, dass die DAG-Parsingzeiten von entwickelten DAGs nicht zu stark verlängert werden.

  9. Airflow-Aufgaben können aus mehreren Gründen fehlschlagen. Damit Fehler bei der Ausführung ganzer DAGs vermieden werden, sollten Sie das Wiederholen von 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 als 0 für die Aufgabe zu überschreiben. Darüber hinaus können Sie den Parameter retries auf Aufgabenebene festlegen.

  10. Wenn Sie GPU in Ihren Airflow-Aufgaben verwenden möchten, erstellen Sie einen separaten GKE-Cluster anhand von Knoten, die Maschinen mit GPUs verwenden. Verwenden Sie den GKEStartPodOperator, um Ihre Aufgaben auszuführen.

  11. Vermeiden Sie es, CPU- und arbeitsspeicherintensive Aufgaben im Knotenpool des Clusters auszuführen, in dem andere Airflow-Komponenten (Planer, Worker, Webserver) ausgeführt werden. Verwenden Sie stattdessen KubernetesPodOperator oder GKEStartPodOperator.

  12. Laden Sie beim Bereitstellen von DAGs in einer Umgebung nur die Dateien in den Ordner /dags hoch, die zum Interpretieren und Ausführen von DAGs unbedingt erforderlich sind.

  13. Begrenzen Sie die Anzahl der DAG-Dateien im Ordner /dags.

    Airflow parst kontinuierlich DAGs im Ordner /dags. Beim Parsen wird der DAGs-Ordner in einer Schleife durchlaufen. Die Anzahl der Dateien, die (mit ihren Abhängigkeiten) geladen werden müssen, wirkt sich auf die Leistung des DAG-Parsings und der Aufgabenplanung aus. Es ist wesentlich effizienter, 100 Dateien mit jeweils 100 DAGs zu verwenden als 10.000 Dateien mit jeweils 1 DAG. Daher wird eine solche Optimierung empfohlen. Diese Optimierung ist ein ausgewogenes Verhältnis zwischen Parsing-Zeit und Effizienz beim Erstellen und Verwalten von DAGs.

    Wenn Sie beispielsweise 10.000 DAG-Dateien bereitstellen möchten, können Sie auch 100 ZIP-Dateien mit jeweils 100 DAG-Dateien erstellen.

    Zusätzlich zu den obigen Hinweisen ist es möglicherweise eine gute Option, DAGs programmgesteuert zu generieren, wenn Sie mehr als 10.000 DAG-Dateien haben. Beispielsweise können Sie eine einzelne Python-DAG-Datei implementieren, die eine bestimmte Anzahl von DAG-Objekten generiert (z. B. 20.100 DAG-Objekte).

Veraltete Airflow-Operatoren vermeiden

Die in der folgenden Tabelle aufgeführten Operatoren wurden eingestellt. Verwenden Sie sie nicht in Ihren DAGs. Verwenden Sie stattdessen aktuelle Alternativen.

Verworfener 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 zu definieren, um Codewiederholungen zu minimieren.

Wie kann ich Code in mehreren DAG-Dateien wiederverwenden?

Speichern Sie Dienstprogrammfunktionen in einer lokalen Python-Bibliothek und importieren Sie 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 DAG A ausgelöst werden soll, können Sie einen TriggerDagRunOperator an das Ende von DAG A setzen.

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 in XComs abgelegt, sodass Sie in anderen Operatoren über Vorlagenfelder darauf verweisen können.

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 dazu Airflow-Pools in der Airflow-Weboberfläche definieren und in Ihren DAGs Aufgaben mit vorhandenen Pools verknüpfen.

FAQs zur Verwendung von Operatoren

Soll ich den DockerOperator verwenden?

Die Verwendung des DockerOperator wird nicht empfohlen, es sei denn, er wird verwendet, um Container in einer Remote-Docker-Installation (nicht innerhalb eines Clusters einer Umgebung) zu starten. 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-Clustern. Wir raten davon ab, Pods im Cluster einer Umgebung zu starten, da dies zu einem Ressourcenwettbewerb führen kann.

Soll ich den SubDagOperator verwenden?

Die Verwendung von SubDagOperator wird nicht empfohlen.

Verwenden Sie Alternativen, wie in der Anleitung zum Gruppieren von Aufgaben empfohlen.

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.

Du kannst die KubernetesPodOperator verwenden. Mit diesem Operator können Sie Kubernetes-Pods definieren und die Pods in anderen Clustern ausfü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,

Sie können auch den KubernetesPodOperator verwenden, um einen Kubernetes-Pod mit einem eigenen Image auszuführen, das Sie aus benutzerdefinierten Paketen erstellt haben.

Wie übergebe ich Argumente einheitlich an einen DAG und die zugehörigen Aufgaben?

Sie können die integrierte Airflow-Unterstützung für Jinja-Vorlagen nutzen, um Argumente zu übergeben, die in Vorlagenfeldern verwendet werden können.

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 dies, dass Vorlagen erst kurz vor der Ausführung einer Aufgabe ersetzt werden.

Wie kann ich erkennen, welche Operatorargumente die Vorlagenersetzung unterstützen?

Operatorargumente, die die Jinja2-Vorlagenersetzung unterstützen, werden explizit entsprechend gekennzeichnet.

Suchen Sie in der Operatordefinition nach dem Feld template_fields. Es enthält eine Liste der Argumentnamen, die einer Vorlagenersetzung unterzogen werden.

Dazu gehört beispielsweise der BashOperator, mit dem Vorlagen für die Argumente bash_command und env unterstützt werden.

Nächste Schritte