Cloud Composer 1 | Cloud Composer 2 | Cloud Composer 3
Auf dieser Seite wird erläutert, wie Sie DAGs, Daten und Konfigurationen aus Ihren vorhandenen Airflow 1.10.*-Umgebungen in Umgebungen mit Airflow 2 und höheren Versionen übertragen.
Weitere Migrationsleitfäden
Von | Bis | Methode | Leitfaden |
---|---|---|---|
Cloud Composer 1, Airflow 2 | Cloud Composer 2, Airflow 2 | Im Vergleich, mit Snapshots | Migrationsanleitung (Snapshots) |
Cloud Composer 1, Airflow 1 | Cloud Composer 2, Airflow 2 | Im Vergleich, mit Snapshots | Migrationsleitfaden (Snapshots) |
Cloud Composer 1, Airflow 2 | Cloud Composer 2, Airflow 2 | Manuelle Übertragung nebeneinander | Manuelle Migration |
Cloud Composer 1, Airflow 1 | Cloud Composer 2, Airflow 2 | Parallele, manuelle Übertragung | Anleitung zur manuellen Migration |
Airflow 1 | Airflow 2 | Manuelle Übertragung nebeneinander | Dieses Handbuch (manuelle Migration) |
Gegenseitige Upgrades
Cloud Composer stellt das Skript der Cloud Composer-Datenbankübertragung zum Migrieren der Metadatendatenbank, DAGs, Daten und Plug-ins aus Cloud Composer-Umgebungen mit Airflow 1.10.14 und Airflow 1.10.15 bereit auf vorhandene Cloud Composer-Umgebungen mit Airflow 2.0.1 und höheren Airflow-Versionen.
Dies ist ein alternativer Pfad zu dem in dieser Anleitung beschriebenen Pfad. Einige Teile dieser Anleitung gelten weiterhin, wenn Sie das bereitgestellte Skript verwenden. Beispielsweise können Sie vor der Migration Ihre DAGs auf Kompatibilität mit Airflow 2 prüfen oder dafür sorgen, dass gleichzeitige DAG-Ausführungen nicht erfolgen und es keine zusätzlichen oder fehlenden DAG-Ausführungen gibt.
Hinweis
Bevor Sie Cloud Composer-Umgebungen mit Airflow 2 verwenden, sollten Sie Änderungen berücksichtigen, die von Airflow 2 in Cloud Composer-Umgebungen übernommen werden.
Planer-HA
Sie können in Ihrer Umgebung mehrere Airflow-Planer verwenden. Sie können die Anzahl der Planer festlegen, wenn Sie eine Umgebung erstellen oder eine vorhandene Umgebung aktualisieren.
Celery+Kubernetes-Executor
Airflow 2 Celery+Kubernetes Executor wird in Cloud Composer 3 unterstützt.
Wichtige Änderungen
Airflow 2 führt viele wichtige Änderungen ein, von denen einige nicht funktionieren.
- Vorhandene DAGs aus Airflow 1.10.* funktionieren möglicherweise nicht mit Airflow 2. Sie müssen getestet und möglicherweise angepasst werden.
- Operatoren, Übertragungen und Hooks, die zu Anbieterpaketen migriert wurden. Importanweisungen in DAGs müssen neue Anbieterpakete verwenden. Alte Importanweisungen funktionieren möglicherweise in Airflow 2 nicht mehr.
- Einige Airflow 1.10.*-Konfigurationen werden möglicherweise nicht mehr unterstützt, da 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 mit Zugriffssteuerung ist die Standard-UI von Airflow 2. Airflow 2 unterstützt keine anderen Airflow-UI-Typen.
- Die experimentelle REST API wird durch die stabile Airflow API ersetzt. Die experimentelle REST API ist in Airflow 2 standardmäßig deaktiviert.
- Andere wichtige Änderungen in Airflow 2.0.0
- Andere 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:
- Umgebungen mit Airflow 2 verwenden Python 3.8. Dies ist eine neuere Version als die, die in Airflow 1.10.*-Umgebungen verwendet wird. Python 2, Python 3.6, Python 3.7 werden nicht unterstützt.
- Airflow 2 verwendet ein anderes Befehlszeilenformat. Cloud Composer unterstützt das neue Format in Umgebungen mit Airflow 2 über den Befehl
gcloud composer environments run
. - Vorinstallierte PyPI-Pakete sind in Airflow 2-Umgebungen anders. Eine Liste der vorinstallierten PyPI-Pakete finden Sie unter 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 und Versuche, sie 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 Airflow-Standardnutzerrolle
Op
. Bei Umgebungen mit Airflow 1.10.* ist die StandardrolleAdmin
.
Schritt 1: Kompatibilität mit Airflow 2 prüfen
Informationen zur Prüfung auf potenzielle Konflikte mit Airflow 2 finden Sie im Anleitung zu Airflow 2.0 und höher, im Abschnitt über Upgrades von DAGs durchführen
Ein häufig auftretendes Problem hängt mit einem inkompatiblen Import zusammen. Pfaden. Weitere Informationen zur Behebung dieses Kompatibilitätsproblems finden Sie in der Anleitung zum Upgrade auf Airflow 2.0 oder höher im Abschnitt zu Backport-Anbietern.
Schritt 2: Airflow 2-Umgebung erstellen, Konfigurationskonfigurationen überschreiben und Umgebungsvariablen
Erstellen Sie eine Airflow 2-Umgebung und Überschreibungen der Übertragungskonfiguration und Umgebungsvariablen:
Führen Sie die Schritte zum Erstellen einer Umgebung aus. Bevor Sie eine Umgebung erstellen, geben Sie auch Konfigurationsüberschreibungen und Umgebungsvariablen an, wie später erläutert.
Wählen Sie unter Image mit Airflow 2 ein Image aus.
Übertragen Sie Konfigurationsparameter manuell von Ihrer Airflow 1.10.*-Umgebung in die neue Airflow 2-Umgebung.
Console
Maximieren Sie beim Erstellen einer Umgebung den Bereich Netzwerk, Airflow-Konfigurationsüberschreibungen und zusätzliche Features.
Klicken Sie unter Airflow-Konfigurationsüberschreibungen auf Airflow-Konfigurationsüberschreibung hinzufügen.
Kopieren Sie alle Konfigurationsüberschreibungen aus Ihrer Airflow 1.10.*-Umgebung.
Einige Konfigurationsoptionen verwenden in Airflow 2 einen anderen Namen und Abschnitt. Weitere Informationen finden Sie unter Konfigurationsänderungen.
Klicken Sie unter Umgebungsvariablen auf Variable hinzufügen.
Kopieren Sie alle Umgebungsvariablen aus Ihrer Airflow 1.10.*-Umgebung.
Klicken Sie auf Erstellen, um eine Umgebung zu erstellen.
Schritt 3: PyPI-Pakete in der Airflow 2-Umgebung installieren
Installieren Sie in der erstellten Airflow 2-Umgebung PyPI-Pakete:
Console
Rufen Sie in der Google Cloud Console die Seite Umgebungen auf.
Wählen Sie die Airflow 2-Umgebung aus.
Wechseln Sie zum Tab PyPI-Pakete und klicken Sie auf Bearbeiten.
Kopieren Sie die PyPI-Paketanforderungen aus Ihrer Airflow 1.10.*-Umgebung. Klicken Sie auf Speichern und warten Sie, bis die Umgebung aktualisiert wurde.
Da Airflow 2-Umgebungen eine andere Reihe von vorinstallierten Paketen und eine andere Python-Version verwenden, können PyPI-Paketkonflikte auftreten, die schwer zu beheben sind. Eine Möglichkeit zur Diagnose von Problemen mit der Paketabhängigkeit besteht darin, eine Prüfung auf PyPI-Paketfehler durchzuführen. Installieren Sie dazu Pakete in einem Airflow-Worker-Pod.
Schritt 4: Variablen und Pools an Airflow 2 übertragen
Airflow 1.10.* unterstützt den Export von Variablen und Pools in JSON-Dateien. Sie können diese Dateien dann in Ihre Airflow 2-Umgebung importieren.
Sie müssen Pools nur übertragen, wenn Sie andere benutzerdefinierte Pools als default_pool
haben. Andernfalls können Sie Befehle überspringen, mit denen Pools exportiert und importiert werden.
gcloud
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 Ihrer Airflow 1.10-Umgebung.AIRFLOW_1_LOCATION
durch die Region, in der sich die Umgebung befindet.
Exportieren Sie Pools aus der Airflow 1.10.*-Umgebung:
gcloud composer environments run AIRFLOW_1_ENV \ --location AIRFLOW_1_LOCATION \ pool -- -e /home/airflow/gcs/data/pools.json
Rufen Sie den URI des Airflow 2-Umgebungs-Buckets ab.
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 Ihrer Airflow 2-Umgebung.AIRFLOW_2_LOCATION
durch die Region, in der sich die Umgebung befindet.
Entfernen Sie in der Ausgabe den Ordner
/dags
. Das Ergebnis ist der URI von Ihren Airflow 2-Umgebungs-Bucket.Ändern Sie beispielsweise
gs://us-central1-example-916807e1-bucket/dags
ings://us-central1-example-916807e1-bucket
.
Ü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 Ihres Airflow 2-Umgebungs-Buckets, den Sie im vorherigen Schritt erhalten haben.Importieren Sie Variablen und Pools in Airflow 2:
gcloud composer environments run \ AIRFLOW_2_ENV \ --location AIRFLOW_2_LOCATION \ variables import \ -- /home/airflow/gcs/data/variables.json gcloud composer environments run \ AIRFLOW_2_ENV \ --location AIRFLOW_2_LOCATION \ pools import \ -- /home/airflow/gcs/data/pools.json
Prüfen Sie, ob Variablen und Pools importiert werden:
gcloud composer environments run \ AIRFLOW_2_ENV \ --location AIRFLOW_2_LOCATION \ variables list gcloud composer environments run \ AIRFLOW_2_ENV \ --location AIRFLOW_2_LOCATION \ pools list
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 1.10.*-Umgebungs-Bucket übertragen
gcloud
Übertragen Sie Plug-ins in Ihre Airflow 2-Umgebung. Exportieren Sie dazu Plug-ins aus Ihrem Airflow 1.10.*-Umgebungs-Bucket in den Ordner
/plugins
Ihres Airflow 2-Umgebungs-Buckets:gcloud composer environments storage plugins export \ --destination=AIRFLOW_2_BUCKET/plugins \ --environment=AIRFLOW_1_ENV \ --location=AIRFLOW_1_LOCATION
Prüfen Sie, ob der Ordner
/plugins
erfolgreich importiert wurde:gcloud composer environments storage plugins list \ --environment=AIRFLOW_2_ENV \ --location=AIRFLOW_2_LOCATION
Exportieren Sie den Ordner
/data
aus Ihrer 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
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 den Export von Nutzern und Verbindungen nicht. Zum Übertragen neue Nutzerkonten und Verbindungen manuell erstellen, Ihre Airflow 2-Umgebung.
gcloud
Führen Sie den 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
Um eine neue Verbindung in Ihrer Airflow 2-Umgebung zu erstellen, führen Sie den
connections
-Airflow-Befehlszeile über gcloud. Beispiel:gcloud 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
So rufen Sie eine Liste der Nutzer in Ihrer Airflow 1.10-Umgebung auf:
Öffnen Sie die Airflow-Weboberfläche für Ihre Airflow 1.10.*-Umgebung.
Klicken Sie auf Admin > Nutzer.
Führen Sie den Befehl
users create
der Airflow-Befehlszeile über gcloud aus, um ein neues Nutzerkonto in Ihrer Airflow 2-Umgebung zu erstellen. Beispiel:gcloud 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 \ --use-random-password \ --role Admin
Schritt 7: Prüfen Sie, ob Ihre DAGs für Airflow 2 bereit sind.
Achten Sie vor der Übertragung von DAGs an Ihre Airflow 2-Umgebung auf Folgendes:
Die DAGs werden erfolgreich ausgeführt und es gibt keine verbleibenden Kompatibilitätsprobleme.
Ihre DAGs verwenden die richtigen Importanweisungen.
Die neue Importanweisung für
BigQueryCreateDataTransferOperator
kann beispielsweise so aussehen:from airflow.providers.google.cloud.operators.bigquery_dts \ import BigQueryCreateDataTransferOperator
Ihre DAGs werden für Airflow 2 aktualisiert. Diese Änderung ist mit Airflow 1.10.14 und höheren 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 aktiviert (nicht pausiert) ist, führt jede Umgebung wie geplant eine eigene Kopie des DAG aus. Dies kann zu gleichzeitigen DAG-Ausführungen für dieselben Daten und zu den gleichen Ausführungszeiten führen.
Aufgrund des DAG-Catchups plant Airflow zusätzliche DAG-Ausführungen, beginnend ab dem in Ihren DAGs angegebenen Startdatum. Dies liegt daran, dass die neue Airflow-Instanz den Verlauf von DAG-Ausführungen aus der Umgebung 1.10.* nicht berücksichtigt. Dies kann zu einer großen Anzahl von geplanten DAG-Ausführungen ab dem angegebenen Startdatum führen.
Gleichzeitige DAG-Ausführungen verhindern
In Ihrer Airflow 2-Umgebung überschreiben Sie die Airflow-Konfigurationsoption dags_are_paused_at_creation
. Nachdem Sie diese Änderung vorgenommen haben, 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, das Sie in Ihre Airflow 2-Umgebung übertragen.
Zur Vermeidung von Lücken und Überschneidungen in Ausführungsdaten sollte die erste DAG-Ausführung in der Airflow 2-Umgebung beim nächsten Auftreten des Zeitplanintervalls erfolgen. 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 15:00 Uhr, 17:00 Uhr und 21:00 Uhr in der Airflow-1.10-Umgebung ausgeführt wird, die letzte DAG-Ausführung um 15:00 Uhr ausgeführt wurde und Sie möchten den DAG um 15:15 Uhr übertragen, dann kann das Startdatum für die Airflow 2-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.
Ein weiteres Beispiel: Wenn Ihr DAG täglich in der Airflow 1.10-Umgebung um 00:00 Uhr ausgeführt wird, die letzte DAG-Ausführung am 26. April 2021 um 00:00 Uhr ausgeführt wurde und Sie möchten den DAG um 13:00 Uhr am 26. April 2021 übertragen, kann das Startdatum für die Airflow 2-Umgebung 23:45 Uhr am 25. April 2021 sein. Nachdem Sie den DAG in der Airflow 2-Umgebung aktiviert haben, plant Airflow einen DAG-Ausführung für 00:00 Uhr am 27. April 2021.
DAGs einzeln in die Airflow 2-Umgebung übertragen
Führen Sie für jeden DAG die folgenden Schritte aus, um ihn zu übertragen:
Achten Sie darauf, dass das neue Startdatum im DAG wie im vorherigen Abschnitt beschrieben festgelegt ist.
Laden Sie den aktualisierten DAG in die Airflow 2-Umgebung hoch. Dieser DAG wird aufgrund der Konfigurationsüberschreibung in der Airflow 2-Umgebung pausiert, sodass keine DAG-Ausführungen geplant sind.
Wechseln Sie in der Airflow-Weboberfläche zu DAGs und suchen Sie nach gemeldeten DAG-Syntaxfehlern.
Zu dem Zeitpunkt, an dem Sie den DAG übertragen möchten:
Halten Sie den DAG in Ihrer Airflow 1.10.*-Umgebung an.
Deaktivieren Sie den DAG in Ihrer Airflow 2-Umgebung.
Prüfen Sie, ob die neue DAG-Ausführung zum richtigen Zeitpunkt geplant ist.
Warten Sie, bis die DAG-Ausführung in der Airflow 2-Umgebung erfolgt ist, und prüfen Sie, ob die Ausführung erfolgreich ist.
Je nachdem, ob die DAG-Ausführung erfolgreich ist:
Wenn die DAG-Ausführung erfolgreich ist, können Sie fortfahren und den DAG aus Ihrer Airflow 2-Umgebung verwenden. Erwägen Sie schließlich, die Airflow-Version 1.10.* des DAG zu löschen.
Wenn die DAG-Ausführung fehlgeschlagen ist, versuchen Sie, die Fehlerbehebung beim DAG durchzuführen, bis sie erfolgreich in Airflow 2 ausgeführt wird.
Bei Bedarf können Sie jederzeit auf die Airflow 1.10.*-Version des DAG zurückgreifen:
Halten Sie den DAG in Ihrer Airflow 2-Umgebung an.
Heben Sie das Pausieren des DAG in Ihrer Airflow 1.10.*-Umgebung auf. Dadurch wird eine neue DAG-Ausführung für das gleiche Datum und die gleiche Uhrzeit wie die fehlgeschlagene DAG-Ausführung geplant.
Wenn Sie mit der Airflow 2-Version des DAG fortfahren möchten, 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 Konfigurationen in die Airflow 2-Umgebung übertragen haben, überwachen Sie diese auf potenzielle Probleme, fehlgeschlagene DAG-Ausführungen und den allgemeinen Umgebungsstatus. Wenn die Airflow 2-Umgebung für einen ausreichend langen Zeitraum fehlerfrei ausgeführt wird, können Sie die Airflow 1.10*-Umgebung entfernen.
Nächste Schritte
- Fehlerbehebung bei DAGs
- Fehlerbehebung beim Erstellen der Umgebung
- Fehlerbehebung bei Umgebungsupdates
- Backport-Pakete verwenden