Umgebungen zu Airflow 2 migrieren

Auf dieser Seite wird erläutert, wie Sie DAGs, Daten und Konfiguration aus vorhandenen Airflow 1.10-Umgebungen in Umgebungen mit Airflow 2 und höher übertragen.

Hinweis

Bevor Sie Cloud Composer-Umgebungen mit Airflow 2 verwenden, sollten Sie sich überlegen, welche Änderungen Airflow für Cloud Composer-Umgebungen bietet.

Planer-HA

Airflow 2 (Planer für hohe Verfügbarkeit) wird in der aktuellen Version von Cloud Composer nicht unterstützt.

Celery+Kubernetes Executor

Airflow 2 Celery+Kubernetes Executor wird in der aktuellen Version von Cloud Composer nicht unterstützt.

Wichtige Änderungen

Airflow 2 führt viele wichtige Änderungen durch:

  • Es kann nicht garantiert werden, dass vorhandene DAGs ab Airflow 1.10* mit Airflow 2 funktionieren. Sie müssen getestet und gegebenenfalls angepasst werden.
  • Operatoren, Übertragungen, Hooks, die zu Anbieterpaketen migriert wurden. Importanweisungen in DAGs müssen neue Anbieterpakete verwenden. Alte Importanweisungen funktionieren in Airflow 2 möglicherweise nicht mehr.
  • Einige Airflow 1.10*-Konfigurationen werden möglicherweise nicht mehr unterstützt, weil Airflow 2 bestimmte Konfigurationsoptionen nicht mehr unterstützt.
  • Einige benutzerdefinierte PyPI-Pakete sind möglicherweise nicht mit der neuen Version von Airflow oder Python kompatibel.
  • Die Airflow-UI für die rollenbasierte Zugriffssteuerung ist die Standard-UI der Airflow 2. Airflow 2 unterstützt keine anderen Airflow-UI-Typen.
  • Die experimentelle REST API wird durch die stable Airflow API ersetzt. Die experimentelle REST API ist in Airflow 2 standardmäßig deaktiviert.
  • Weitere wichtige Änderungen in Airflow 2.0.0
  • Weitere wichtige Änderungen in Airflow 2.0.1

Unterschiede zwischen Umgebungen mit Airflow 2 und Airflow 1.10.*

Hauptunterschiede zwischen Cloud Composer-Umgebungen mit Airflow 1.10.* und Umgebungen mit Airflow 2:

  • Für Umgebungen mit Airflow 2 wird Python 3.8 verwendet. Dies ist eine neuere Version als die in Airflow 1.10.*-Umgebungen. Python 2, Python 3.6 und Python 3.7 werden nicht unterstützt.
  • Airflow 2 verwendet ein anderes CLI-Format. Cloud Composer unterstützt das neue Format in Umgebungen mit Airflow 2 über den Befehl gcloud beta composer environments run.
  • vorinstallierte PyPI-Pakete unterscheiden sich in Airflow 2-Umgebungen. Eine Liste vorinstallierter PyPI-Pakete finden Sie in der Liste der Cloud Composer-Versionen.
  • Die DAG-Serialisierung ist in Airflow 2 immer aktiviert. Daher wird das asynchrone Laden von DAGs nicht mehr benötigt und in Airflow 2 nicht unterstützt. Daher wird die Konfiguration der Parameter [core]store_serialized_dags und [core]store_dag_code für Airflow 2 nicht unterstützt. Versuche, diese festzulegen, werden als Fehler gemeldet.
  • Airflow-Webserver-Plug-ins werden nicht unterstützt. Dies hat keine Auswirkungen auf Planer- oder Worker-Plug-ins, einschließlich Airflow-Operatoren und -Sensoren.
  • In Airflow 2-Umgebungen ist die Standard-RBAC-Nutzerrolle Op. Für Umgebungen mit Airflow 1.10.* lautet die Standardrolle Admin.

Schritt 1: Kompatibilität mit Airflow 2 prüfen

Verwenden Sie zur Prüfung auf mögliche Konflikte mit Airflow 2 in Ihrer bestehenden Airflow 1.10-Umgebung Code-Upgradeskripts.

gcloud

  1. Wenn in Ihrer Umgebung Airflow 1.10.14 und frühere Versionen verwendet werden, führen Sie ein Upgrade Ihrer Umgebung auf eine Cloud Composer-Version mit Airflow 1.10.15 und höher durch. Cloud Composer unterstützt Befehle für die Upgradeprüfung ab Airflow 1.10.15.

  2. Führen Sie mit dem Befehl gcloud beta composer environments run eine Upgrade-Prüfung durch. Einige Upgrade-Prüfungen, die für eigenständiges Airflow 1.10.15 relevant sind, sind für Cloud Composer nicht relevant. Der folgende Befehl schließt diese Prüfungen aus.

    gcloud beta composer environments run \
        AIRFLOW_1_ENV  \
        --location=AIRFLOW_1_LOCATION \
        upgrade_check \
        -- --ignore VersionCheckRule --ignore LoggingConfigurationRule \
        --ignore PodTemplateFileRule --ignore SendGridEmailerMovedRule
    

    Ersetzen Sie:

    • AIRFLOW_1_ENV durch den Namen der Airflow 1.10-Umgebung.
    • AIRFLOW_1_LOCATION durch die Compute Engine-Region, in der sich die Umgebung befindet.
  3. Überprüfen Sie die Ausgabe des Befehls. Aktualisierungsskripts melden potenzielle Kompatibilitätsprobleme in vorhandenen Umgebungen.

  4. Implementieren Sie andere Änderungen an DAGs, wie in der Anleitung "Upgrade auf Airflow 2.0+" im Abschnitt Upgrade von DAGs ausführen beschrieben.

Schritt 2: Airflow-Umgebung erstellen, Konfigurationsüberschreibungen und Umgebungsvariablen übertragen

Erstellen Sie eine Airflow 2-Umgebung und übertragen Sie Konfigurationsüberschreibungen und Umgebungsvariablen:

  1. Führen Sie die Schritte zum Erstellen einer Umgebung aus. Bevor Sie eine Umgebung erstellen, geben Sie auch die Konfigurations-Überschreibungen und Umgebungsvariablen an, wie weiter unten erläutert.

  2. Wählen Sie ein Image aus und wählen Sie ein Image mit Airflow 2 aus.

  3. Konfigurationsparameter aus der Airflow 1.10-Umgebung manuell in die neue Airflow 2-Umgebung übertragen.

    Console

    1. Erweitern Sie beim Erstellen einer Umgebung den Abschnitt Netzwerk, Airflow-Konfigurationsüberschreibungen und zusätzliche Funktionen.

    2. Klicken Sie unter Airflow-Konfigurationsüberschreibungen auf Airflow-Konfigurationsüberschreibung hinzufügen.

    3. Kopieren Sie alle Konfigurationsüberschreibungen aus der Airflow 1.10-Umgebung.

      Einige Konfigurationsoptionen verwenden in Airflow 2 einen anderen Namen und einen anderen Abschnitt. Weitere Informationen finden Sie unter Konfigurationsänderungen.

    4. Klicken Sie unter Umgebungsvariablen auf Umgebungsvariable hinzufügen.

    5. Kopieren Sie alle Umgebungsvariablen aus Ihrer Airflow 1.10-Umgebung.

    6. Klicken Sie auf Erstellen, um eine Umgebung zu erstellen.

Schritt 3: PyPI-Pakete in der Airflow 2-Umgebung installieren

Nachdem Ihre Airflow 2-Umgebung erstellt wurde, installieren Sie die PyPI-Pakete:

Console

  1. Rufen Sie in der Google Cloud Console die Seite Umgebungen auf.

    Zur Seite "Umgebungen"

  2. Wählen Sie Ihre Airflow 2-Umgebung aus.

  3. Klicken Sie auf den Tab PyPI-Pakete und dann auf Bearbeiten.

  4. Kopieren Sie die PyPI-Paketanforderungen aus der Airflow 1.10-Umgebung. Klicken Sie auf Speichern und warten Sie, bis die Umgebung aktualisiert wurde.

    Da für Airflow-2-Umgebungen ein anderer vorinstallierter Pakete und eine andere Python-Version verwendet werden, treten möglicherweise PyPI-Paketkonflikte auf, die schwer zu lösen sind. Eine Möglichkeit, Paketabhängigkeiten zu ermitteln, besteht darin, nach PyPI-Paketfehlern zu suchen, indem Sie Pakete in einem Airflow-Worker-Pod installieren.

Schritt 4: Variablen und Pools an Airflow 2 übertragen

Airflow 1.10.* unterstützt das Exportieren von Variablen und Pools in JSON-Dateien. Anschließend können Sie diese Dateien in Ihre Airflow 2-Umgebung importieren.

Sie müssen Pools nur übertragen, wenn Sie benutzerdefinierte Pools als default_pool haben. Andernfalls überspringen Sie Befehle, die Pools exportieren und importieren.

gcloud

  1. Exportieren Sie Variablen aus Ihrer Airflow 1.10-Umgebung.*

    gcloud composer environments run AIRFLOW_1_ENV \
        --location AIRFLOW_1_LOCATION \
         variables -- -e /home/airflow/gcs/data/variables.json
    

    Ersetzen Sie:

    • AIRFLOW_1_ENV durch den Namen der Airflow 1.10-Umgebung.
    • AIRFLOW_1_LOCATION durch die Compute Engine-Region, in der sich die Umgebung befindet.
  2. Exportieren Sie Pools aus Ihrer Airflow 1.10-Umgebung.*

    gcloud composer environments run AIRFLOW_1_ENV \
        --location AIRFLOW_1_LOCATION \
         pool -- -e /home/airflow/gcs/data/pools.json
    
  3. Rufen Sie den Bucket-URI für die Airflow 2-Umgebung ab.

    1. Führen Sie dazu diesen Befehl aus:

      gcloud composer environments describe AIRFLOW_2_ENV \
          --location AIRFLOW_2_LOCATION \
           --format="value(config.dagGcsPrefix)"
      

      Ersetzen Sie:

      • AIRFLOW_2_ENV durch den Namen der Airflow 2-Umgebung.
      • AIRFLOW_2_LOCATION durch die Compute Engine-Region, in der sich die Umgebung befindet.
    2. Entfernen Sie in der Ausgabe den Ordner /dags. Das Ergebnis ist der URI Ihres Airflow 1.10-Umgebungs-Buckets.

      Ändern Sie beispielsweise gs://us-central1-example-916807e1-bucket/dags in gs://us-central1-example-916807e1-bucket.

  4. Übertragen Sie JSON-Dateien mit Variablen und Pools in Ihre Airflow 2-Umgebung:

    gcloud composer environments storage data export \
        --destination=AIRFLOW_2_BUCKET/data \
        --environment=AIRFLOW_1_ENV \
        --location=AIRFLOW_1_LOCATION \
        --source=variables.json
    
    gcloud composer environments storage data export \
        --destination=AIRFLOW_2_BUCKET/data \
        --environment=AIRFLOW_1_ENV \
        --location=AIRFLOW_1_LOCATION \
        --source=pools.json
    

    Ersetzen Sie AIRFLOW_2_BUCKET durch den URI des Airflow-2-Umgebungs-Buckets aus dem vorherigen Schritt.

  5. Variablen und Pools in Airflow importieren:

    gcloud beta composer environments run \
        AIRFLOW_2_ENV \
        --location AIRFLOW_2_LOCATION \
        variables import \
        -- /home/airflow/gcs/data/variables.json
    
    gcloud beta composer environments run \
        AIRFLOW_2_ENV \
        --location AIRFLOW_2_LOCATION \
        pools import \
        -- /home/airflow/gcs/data/pools.json
    
  6. Prüfen Sie, ob Variablen und Pools importiert werden:

    gcloud beta composer environments run \
        AIRFLOW_2_ENV \
        --location AIRFLOW_2_LOCATION \
        variables list
    
    gcloud beta composer environments run \
        AIRFLOW_2_ENV \
        --location AIRFLOW_2_LOCATION \
        pools list
    
  7. Entfernen Sie JSON-Dateien aus den Buckets:

    gcloud composer environments storage data delete \
        variables.json \
        --environment=AIRFLOW_2_ENV \
        --location=AIRFLOW_2_LOCATION
    
    gcloud composer environments storage data delete \
        pools.json \
        --environment=AIRFLOW_2_ENV \
        --location=AIRFLOW_2_LOCATION
    
    gcloud composer environments storage data delete \
        variables.json \
        --environment=AIRFLOW_1_ENV \
        --location=AIRFLOW_1_LOCATION
    
    gcloud composer environments storage data delete \
        pools.json \
        --environment=AIRFLOW_1_ENV \
        --location=AIRFLOW_1_LOCATION
    

Schritt 5: Andere Daten aus Ihrem Airflow-Bucket 1.10.* übertragen

gcloud

  1. Plug-ins in Ihre Airflow 2-Umgebung übertragen Exportieren Sie dazu Plug-ins aus Ihrem Airflow 1.10-Umgebungs-Bucket in den Ordner /plugins in Ihrem Airflow 2-Umgebungs-Bucket:

    gcloud composer environments storage plugins export \
        --destination=AIRFLOW_2_BUCKET/plugins \
        --environment=AIRFLOW_1_ENV \
        --location=AIRFLOW_1_LOCATION
    
  2. Prüfen Sie, ob der Ordner /plugins erfolgreich importiert wurde:

    gcloud composer environments storage plugins list \
        --environment=AIRFLOW_2_ENV \
        --location=AIRFLOW_2_LOCATION
    
  3. Exportieren Sie den Ordner /data aus der Airflow 1.10-Umgebung in die Airflow 2-Umgebung:

        gcloud composer environments storage data export \
            --destination=AIRFLOW_2_BUCKET/data \
            --environment=AIRFLOW_1_ENV \
            --location=AIRFLOW_1_LOCATION
    
  4. Prüfen Sie, ob der Ordner /data erfolgreich importiert wurde:

    gcloud composer environments storage data list \
        --environment=AIRFLOW_2_ENV \
        --location=AIRFLOW_2_LOCATION
    

Schritt 6: Verbindungen und Nutzer übertragen

Airflow 1.10.* unterstützt nicht das Exportieren von Nutzern und Verbindungen. Zum Übertragen von Nutzern und Verbindungen erstellen Sie in der Airflow 1.10-Umgebung manuell neue Nutzerkonten und Verbindungen zu Ihrer Airflow 2-Umgebung.

gcloud

  1. Führen Sie folgenden Befehl aus, um eine Liste der Verbindungen in Ihrer Airflow 1.10-Umgebung* abzurufen:

    gcloud composer environments run AIRFLOW_1_ENV \
        --location AIRFLOW_1_LOCATION \
         connections -- --list
    
  2. Führen Sie den Befehl connections-Airflow-Befehlszeile über gcloud aus, um eine neue Verbindung in Ihrer Airflow 2-Umgebung zu erstellen. Beispiel:

    gcloud beta composer environments run \
        AIRFLOW_2_ENV \
        --location AIRFLOW_2_LOCATION \
        connections add \
        -- --conn-host postgres.example.com \
        --conn-port 5432 \
        --conn-type postgres \
        --conn-login example_user \
        --conn-password example_password \
        --conn-description "Example connection" \
        example_connection
    
  3. So rufen Sie eine Liste der Nutzer in Ihrer Airflow 1.10-Umgebung auf:*

    1. Openffnen Sie die Airflow-Weboberfläche für Ihre Airflow 1.10-Umgebung.*

    2. Klicken Sie auf Verwaltung > Nutzer.

  4. Führen Sie den Befehl users create-Airflow-Befehlszeile über gcloud aus, um ein neues Nutzerkonto in Ihrer Airflow 2-Umgebung zu erstellen. Beispiel:

    gcloud beta composer environments run \
        AIRFLOW_2_ENV \
        --location AIRFLOW_2_LOCATION \
        users create \
        -- --username example_username \
        --firstname Example-Name \
        --lastname Example-Surname \
        --email example-user@example.com \
        --password example_password \
        --role Admin
    

Schritt 7: Prüfen Sie, ob Ihre DAGs für Airflow 2 bereit sind.

Bevor Sie DAGs in Ihre Airflow 2-Umgebung übertragen, stellen Sie sicher, dass:

  1. Upgradeskripts prüft für Ihre DAGs erfolgreich und es gibt keine verbleibenden Kompatibilitätsprobleme.

  2. Ihre DAGs verwenden richtige Importanweisungen.

    Die neue Importanweisung für BigQueryCreateDataTransferOperator kann beispielsweise so aussehen:

    from airflow.providers.google.cloud.operators.bigquery_dts \
        import BigQueryCreateDataTransferOperator
    
  3. Ihre DAGs werden für Airflow 2 aktualisiert. Diese Änderung ist mit Airflow 1.10.14 und späteren Versionen kompatibel.

Schritt 8: DAGs in die Airflow 2-Umgebung übertragen

Die folgenden potenziellen Probleme können auftreten, wenn Sie DAGs zwischen Umgebungen übertragen:

  • Wenn ein DAG in beiden Umgebungen (nicht pausiert) aktiviert ist, wird für jede Umgebung eine eigene Kopie des DAG ausgeführt. Dies kann zu gleichzeitigen DAG-Ausführungen für dieselben Daten und Ausführungszeiten führen.

  • Aufgrund des DAG-Abrufs plant Airflow zusätzliche DAG-Ausführungen ab dem in den DAGs angegebenen Startdatum. Das liegt daran, dass die neue Airflow-Instanz nicht den Verlauf von DAG-Ausführungen aus der Umgebung 1.10.* berücksichtigt. Dies kann zu einer großen Anzahl von DAG-Ausführungen führen, die ab dem angegebenen Startdatum beginnen.

Gleichzeitige DAG-Ausführungen verhindern

Überschreiben Sie die Airflow-Konfigurationsoption dags-are-paused-at-creation in der Airflow 2-Umgebung. Nach dieser Änderung werden alle neuen DAGs standardmäßig pausiert.

Bereich Schlüssel Wert
core dags-are-paused-at-creation True

Zusätzliche oder fehlende DAG-Ausführungen verhindern

Geben Sie ein neues statisches Startdatum in DAGs an, die Sie in Ihre Airflow 2-Umgebung übertragen.

Damit Lücken und Überlappungen in Ausführungsdaten vermieden werden, sollte die erste DAG-Ausführung in der Airflow-Umgebung 2 nach dem nächsten Auftreten des Zeitplanintervalls ausgeführt werden. Legen Sie dazu das neue Startdatum in Ihrem DAG vor dem Datum der letzten Ausführung in der Airflow 1.10-Umgebung fest.*

Wenn Ihr DAG beispielsweise täglich um 17:00 Uhr, 17:00 Uhr und 23:00 Uhr in der Airflow 1.10-Umgebung ausgeführt wird, ist die letzte DAG-Ausführung um 17:00 Uhr. Sie möchten Übertragen Sie den DAG um 15:15 Uhr. Anschließend kann das Startdatum für die Airflow-Umgebung heute um 14:45 Uhr sein. Nachdem Sie den DAG in der Airflow 2-Umgebung aktiviert haben, plant Airflow eine DAG-Ausführung für 17:00 Uhr.

Weitere Informationen: Wenn Ihr DAG an jedem Tag in der Airflow-Umgebung 1.10 um 00:00 Uhr ausgeführt wird.*, wurde der letzte DAG am 26. April 2021 um 00:00 Uhr ausgeführt und Sie möchten den DAG übertragen am 26. April 2021 um 23:00 Uhr ist das Startdatum für die Airflow 2-Umgebung am 25. April 2021 um 23:45 Uhr. Nachdem Sie den DAG in der Airflow 2-Umgebung aktiviert haben, plant Airflow am 27. April 2021 eine DAG-Ausführung für 00:00.

DAGs einzeln in die Airflow 2-Umgebung übertragen

Führen Sie für jeden DAG folgende Schritte zur Übertragung aus:

  1. Sorgen Sie dafür, dass das neue Startdatum im DAG wie im vorherigen Abschnitt beschrieben festgelegt ist.

  2. Laden Sie den aktualisierten DAG in die Airflow 2-Umgebung hoch. Dieser DAG wird in der Airflow-Umgebung 2 aufgrund der Konfigurationsüberschreibung pausiert. In diesem Fall werden noch keine DAG-Ausführungen geplant.

  3. Wechseln Sie in der Airflow-Weboberfläche zu DAGs und suchen Sie nach gemeldeten DAG-Syntaxfehlern.

  4. Wann Sie den DAG übertragen möchten:

    1. Pausieren Sie den DAG in Ihrer Airflow 1.10-Umgebung.*

    2. Heben Sie die Pausierung des DAG in Ihrer Airflow 2-Umgebung auf.

    3. Überprüfen Sie, ob der neue DAG zur richtigen Zeit geplant wird.

    4. Warten Sie, bis die DAG-Ausführung in der Airflow 2-Umgebung ausgeführt wird, und prüfen Sie, ob die Ausführung erfolgreich war.

  5. Je nachdem, ob die DAG-Ausführung erfolgreich ausgeführt wurde:

    • Wenn die DAG-Ausführung erfolgreich ausgeführt wurde, können Sie fortfahren und den DAG aus Ihrer Airflow 2-Umgebung verwenden. Denken Sie darüber nach, die Airflow-Version 1.10* des DAG zu löschen.

    • Wenn die DAG-Ausführung fehlgeschlagen ist, versuchen Sie, das Problem mit dem DAG zu beheben, bis er in Airflow 2 erfolgreich ausgeführt wurde.

      Bei Bedarf können Sie jederzeit zur Airflow-Version 1.10* des DAG zurückkehren:

      1. Pausieren Sie den DAG in Ihrer Airflow 2-Umgebung.

      2. Heben Sie die Pausierung des DAG in Ihrer Airflow 1.10-Umgebung auf. Dadurch wird eine neue DAG-Ausführung für dasselbe Datum und dieselbe Uhrzeit wie der fehlgeschlagene DAG-Ausführung geplant.

      3. Wenn Sie bereit sind, mit der Airflow-Version 2 des DAG fortzufahren, passen Sie das Startdatum an, laden Sie die neue Version des DAG in Ihre Airflow 2-Umgebung hoch und wiederholen Sie den Vorgang.

Schritt 9: Airflow 2-Umgebung überwachen

Nachdem Sie alle DAGs und die Konfiguration in die Airflow 2-Umgebung übertragen haben, sollten Sie sie auf mögliche Probleme, fehlgeschlagene DAG-Ausführungen und den Gesamtzustand der Umgebung überwachen. Wenn die Airflow 2-Umgebung während eines ausreichenden Zeitraums fehlerfrei ausgeführt wird, können Sie die Airflow 1.10-Umgebung entfernen.

Nächste Schritte