Migrazione degli ambienti a Airflow 2

Cloud Composer 3 | Cloud Composer 2 | Cloud Composer 1

Questa pagina spiega come trasferire DAG, dati e configurazione dagli ambienti Airflow 1.10.* esistenti agli ambienti con Airflow 2 e versioni successive di Airflow.

Altre guide alla migrazione

Da A Metodo Guida
Cloud Composer 1, Airflow 2 Cloud Composer 2, Airflow 2 Accanto, utilizzando gli snapshot Guida alla migrazione (snapshot)
Cloud Composer 1, Airflow 1 Cloud Composer 2, Airflow 2 Accanto, utilizzando gli snapshot Guida alla migrazione (snapshot)
Cloud Composer 1, Airflow 2 Cloud Composer 2, Airflow 2 Trasferimento manuale affiancato Guida alla migrazione manuale
Cloud Composer 1, Airflow 1 Cloud Composer 2, Airflow 2 Trasferimento manuale affiancato Guida alla migrazione manuale
Airflow 1 Airflow 2 Trasferimento manuale affiancato Questa guida (migrazione manuale)

Upgrade affiancati

Cloud Composer fornisce lo script di trasferimento del database Cloud Composer per eseguire la migrazione del database dei metadati, degli DAG, dei dati e dei plug-in dagli ambienti Cloud Composer con Airflow 1.10.14 e Airflow 1.10.15 agli ambienti Cloud Composer esistenti con Airflow 2.0.1 e versioni successive di Airflow.

Si tratta di un percorso alternativo rispetto a quello descritto in questa guida. Alcune parti di questa guida si applicano ancora quando si utilizza lo script fornito. Ad esempio, potresti voler verificare la compatibilità dei DAG con Airflow 2 prima di eseguirne la migrazione o assicurarti che non vengano eseguite esecuzioni DAG concorrenti e che non siano presenti esecuzioni DAG extra o mancanti.

Prima di iniziare

Prima di iniziare a utilizzare gli ambienti Cloud Composer con Airflow 2, tieni conto delle modifiche apportate da Airflow 2 agli ambienti Cloud Composer.

Più pianificatori

Puoi utilizzare più di uno scheduler Airflow nel tuo ambiente. Puoi impostare il numero di pianificatori quando crei un ambiente o aggiorni un ambiente esistente.

Esecutore Celery+Kubernetes

L'executor Celery+Kubernetes di Airflow 2 è supportato in Cloud Composer 3.

Modifiche che provocano un errore

Airflow 2 introduce molte modifiche importanti, alcune delle quali non sono compatibili con le versioni precedenti:

Differenze tra gli ambienti con Airflow 2 e Airflow 1.10.*

Differenze principali tra gli ambienti Cloud Composer con Airflow 1.10.* e gli ambienti con Airflow 2:

  • Gli ambienti con Airflow 2 utilizzano Python 3.8. Si tratta di una versione più recente di quella utilizzata negli ambienti Airflow 1.10.*. Python 2, Python 3.6 e Python 3.7 non sono supportati.
  • Airflow 2 utilizza un formato CLI diverso. Cloud Composer supporta il nuovo formato negli ambienti con Airflow 2 tramite il comando gcloud composer environments run.
  • I pacchetti PyPI preinstallati sono diversi negli ambienti Airflow 2. Per un elenco dei pacchetti PyPI preinstallati, consulta l'elenco delle versioni di Cloud Composer.
  • La serializzazione DAG è sempre attiva in Airflow 2. Di conseguenza, il caricamento DAG asincrono non è più necessario e non è supportato in Airflow 2. Di conseguenza, la configurazione dei parametri [core]store_serialized_dags e [core]store_dag_code non è supportata per Airflow 2 e i tentativi di impostarli verranno segnalati come errori.
  • I plug-in del server web Airflow non sono supportati. Ciò non influisce sui plug-in di pianificatori o worker, inclusi gli operatori e i sensori di Airflow.
  • Negli ambienti Airflow 2, il ruolo utente Airflow predefinito è Op. Per gli ambienti con Airflow 1.10.*, il ruolo predefinito è Admin.

Passaggio 1: verifica la compatibilità con Airflow 2

Per verificare la presenza di potenziali conflitti con Airflow 2, consulta la guida sull'upgrade ad Airflow 2.0 e versioni successive, nella sezione relativa all'upgrade dei DAG.

Un problema comune che potresti riscontrare riguarda i percorsi di importazione non compatibili. Per ulteriori informazioni sulla risoluzione di questo problema di compatibilità, consulta la sezione relativa ai provider di backport nella guida all'upgrade ad Airflow 2.0 e versioni successive.

Passaggio 2: crea un ambiente Airflow 2, trasferisci le sostituzioni di configurazione e le variabili di ambiente

Crea un ambiente Airflow 2 e trasferisci le override della configurazione e le variabili di ambiente:

  1. Segui i passaggi per la creazione di un ambiente. Prima di creare un ambiente, specifica anche gli override della configurazione e le variabili di ambiente, come spiegato di seguito.

  2. Quando selezioni un'immagine, scegli un'immagine con Airflow 2.

  3. Trasferisci manualmente i parametri di configurazione dall'ambiente Airflow 1.10.* al nuovo ambiente Airflow 2.

    Console

    1. Quando crei un ambiente, espandi la sezione Networking, override della configurazione di Airflow e funzionalità aggiuntive.

    2. In Override della configurazione di Airflow, fai clic su Aggiungi override della configurazione di Airflow.

    3. Copia tutte le sostituzioni di configurazione dall'ambiente Airflow 1.10.*.

      Alcune opzioni di configurazione utilizzano un nome e una sezione diversi in Airflow 2. Per ulteriori informazioni, consulta Modifiche alla configurazione.

    4. In Variabili di ambiente, fai clic su Aggiungi variabile di ambiente.

    5. Copia tutte le variabili di ambiente dall'ambiente Airflow 1.10.*.

    6. Fai clic su Crea per creare un ambiente.

Passaggio 3: installa i pacchetti PyPI nell'ambiente Airflow 2

Dopo aver creato l'ambiente Airflow 2, installa i pacchetti PyPI:

Console

  1. Nella console Google Cloud, vai alla pagina Ambienti.

    Vai ad Ambienti

  2. Seleziona il tuo ambiente Airflow 2.

  3. Vai alla scheda Pacchetti PyPI e fai clic su Modifica.

  4. Copia i requisiti del pacchetto PyPI dall'ambiente Airflow 1.10.*. Fai clic su Salva e attendi che l'ambiente venga aggiornato.

    Poiché gli ambienti Airflow 2 utilizzano un insieme diverso di pacchetti preinstallati e una versione diversa di Python, potresti riscontrare conflitti di pacchetti PyPI difficili da risolvere.

Passaggio 4: trasferisci le variabili e i pool in Airflow 2

Airflow 1.10.* supporta l'esportazione di variabili e pool in file JSON. Puoi quindi importarli nel tuo ambiente Airflow 2.

Devi trasferire i pool solo se hai pool personalizzati diversi da default_pool. In caso contrario, salta i comandi che esportano e importano i pool.

gcloud

  1. Esporta le variabili dall'ambiente Airflow 1.10.*.

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

    Sostituisci:

    • AIRFLOW_1_ENV con il nome dell'ambiente Airflow 1.10.*.
    • AIRFLOW_1_LOCATION con la regione in cui si trova l'ambiente.

  2. Esporta i pool dal tuo ambiente Airflow 1.10.*:

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

  3. Ottieni l'URI del bucket dell'ambiente Airflow 2.

    1. Esegui questo comando:

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

      Sostituisci:

      • AIRFLOW_2_ENV con il nome del tuo ambiente Airflow 2.
      • AIRFLOW_2_LOCATION con la regione in cui si trova l'ambiente.

    2. Nell'output, rimuovi la cartella /dags. Il risultato è l'URI del tuo bucket dell'ambiente Airflow 2.

      Ad esempio, cambia gs://us-central1-example-916807e1-bucket/dags in gs://us-central1-example-916807e1-bucket.

  4. Trasferisci i file JSON con variabili e pool nell'ambiente Airflow 2:

    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

    Sostituisci AIRFLOW_2_BUCKET con l'URI del bucket dell'ambiente Airflow 2 ottenuto nel passaggio precedente.

  5. Importa le variabili e i pool 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

  6. Verifica che le variabili e i pool siano importati:

    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

  7. Rimuovi i file JSON dai bucket:

    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

Passaggio 5: trasferisci altri dati dal bucket dell'ambiente Airflow 1.10.*

gcloud

  1. Trasferisci i plug-in nell'ambiente Airflow 2. Per farlo, esporta i plug-in dal bucket dell'ambiente Airflow 1.10.* nella cartella /plugins del bucket dell'ambiente Airflow 2:

    gcloud composer environments storage plugins export \
    --destination=AIRFLOW_2_BUCKET/plugins \
    --environment=AIRFLOW_1_ENV \
    --location=AIRFLOW_1_LOCATION

  2. Verifica che la cartella /plugins sia stata importata correttamente:

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

  3. Esporta la cartella /data dall'ambiente Airflow 1.10.* all'ambiente Airflow 2:

        gcloud composer environments storage data export \
        --destination=AIRFLOW_2_BUCKET/data \
        --environment=AIRFLOW_1_ENV \
        --location=AIRFLOW_1_LOCATION

  4. Verifica che la cartella /data sia stata importata correttamente:

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

Passaggio 6: trasferisci connessioni e utenti

Airflow 1.10.* non supporta l'esportazione di utenti e connessioni. Per trasferire gli utenti e le connessioni, crea manualmente nuovi account utente e connessioni nel tuo ambiente Airflow 2.

gcloud

  1. Per ottenere un elenco delle connessioni nell'ambiente Airflow 1.10.*, esegui:

    gcloud composer environments run AIRFLOW_1_ENV \
    --location AIRFLOW_1_LOCATION \
     connections -- --list

  2. Per creare una nuova connessione nel tuo ambiente Airflow 2, esegui il connections comando Airflow CLI tramite gcloud. Ad esempio:

    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

  3. Per visualizzare un elenco di utenti nel tuo ambiente Airflow 1.10.*:

    1. Apri l'interfaccia web di Airflow per il tuo ambiente Airflow 1.10.*.

    2. Vai ad Amministrazione > Utenti.

  4. Per creare un nuovo account utente nel tuo ambiente Airflow 2, esegui il users createcomando Airflow CLI tramite gcloud. Ad esempio:

    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

Passaggio 7: assicurati che i DAG siano pronti per Airflow 2

Prima di trasferire i DAG nell'ambiente Airflow 2, assicurati che:

  1. I DAG vengono eseguiti correttamente e non ci sono altri problemi di compatibilità.

  2. I DAG utilizzino dichiarazioni di importazione corrette.

    Ad esempio, la nuova dichiarazione di importazione per BigQueryCreateDataTransferOperator può avere il seguente aspetto:

    from airflow.providers.google.cloud.operators.bigquery_dts \
    import BigQueryCreateDataTransferOperator

  3. I DAG sono stati sottoposti a upgrade per Airflow 2. Questa modifica è compatibile con Airflow 1.10.14 e versioni successive.

Passaggio 8: trasferisci i DAG nell'ambiente Airflow 2

Quando trasferisci i DAG tra ambienti, potrebbero verificarsi i seguenti potenziali problemi:

  • Se un DAG è attivato (non in pausa) in entrambi gli ambienti, ogni ambiente esegue la propria copia del DAG, come pianificato. Ciò potrebbe comportare l'esecuzione di DAG contemporaneamente per gli stessi dati e per lo stesso momento di esecuzione.

  • A causa del recupero DAG, Airflow pianifica esecuzioni DAG aggiuntive a partire dalla data di inizio specificata nei DAG. Questo accade perché la nuova istanza Airflow non prende in considerazione la cronologia delle esecuzioni DAG dall'ambiente 1.10.*. Ciò potrebbe comportare un numero elevato di esecuzioni pianificate del DAG a partire dalla data di inizio specificata.

Evitare esecuzioni DAG concorrenti

Nell'ambiente Airflow 2, sovrascrivi l'opzione di configurazione di Airflow dags_are_paused_at_creation. Dopo aver apportato questa modifica, tutti i nuovi DAG vengono messi in pausa per impostazione predefinita.

Sezione Chiave Valore
core dags_are_paused_at_creation True

Evitare esecuzioni DAG mancanti o aggiuntive

Specifica una nuova data di inizio statica nei DAG che trasferisci al tuo ambiente Airflow 2.

Per evitare lacune e sovrapposizioni nelle date di esecuzione, la prima esecuzione del DAG deve avvenire nell'ambiente Airflow 2 alla successiva occorrenza dell'intervallo di pianificazione. A tale scopo, imposta la nuova data di inizio nel DAG in modo che sia precedente alla data dell'ultima esecuzione nell'ambiente Airflow 1.10.*.

Ad esempio, se il DAG viene eseguito alle 15:00, alle 17:00 e alle 21:00 ogni giorno nell'ambiente Airflow 1.10.*, l'ultima esecuzione del DAG è avvenuta alle 15:00 e prevedi di trasferire il DAG alle 15:15, la data di inizio per l'ambiente Airflow 2 può essere oggi alle 14:45. Dopo aver attivato il DAG nell'ambiente Airflow 2, Airflow pianifica un'esecuzione del DAG per le 17:00.

Come altro esempio, se il DAG viene eseguito ogni giorno alle 00:00 nell'ambiente Airflow 1.10.*, l'ultima esecuzione del DAG è avvenuta alle 00:00 del 26 aprile 2021 e prevedi di trasferire il DAG alle 13:00 del 26 aprile 2021, la data di inizio per l'ambiente Airflow 2 può essere il 25 aprile 2021 alle 23:45. Dopo aver attivato il DAG nell'ambiente Airflow 2, Airflow pianifica un'esecuzione del DAG per le 00:00 del 27 aprile 2021.

Trasferisci i DAG uno alla volta nell'ambiente Airflow 2

Per ogni DAG, segui questa procedura per trasferirlo:

  1. Assicurati che la nuova data di inizio nel DAG sia impostata come descritto nella sezione precedente.

  2. Carica il DAG aggiornato nell'ambiente Airflow 2. Questo DAG è in pausa nell'ambiente Airflow 2 per via dell'override della configurazione, pertanto non sono ancora pianificate esecuzioni del DAG.

  3. Nell'interfaccia web di Airflow, vai a DAG e controlla se sono stati segnalati errori di sintassi dei DAG.

  4. Al momento in cui prevedi di trasferire il DAG:

    1. Metti in pausa il DAG nell'ambiente Airflow 1.10.*.

    2. Riattiva la messa in pausa del DAG nell'ambiente Airflow 2.

    3. Verifica che l'esecuzione del nuovo DAG sia pianificata all'ora corretta.

    4. Attendi l'esecuzione del DAG nell'ambiente Airflow 2 e controlla se l'esecuzione è andata a buon fine.

  5. A seconda che l'esecuzione del DAG sia riuscita:

    • Se l'esecuzione del DAG è riuscita, puoi procedere e utilizzare il DAG dal tuo ambiente Airflow 2. Infine, valuta la possibilità di eliminare la versione 1.10.* del DAG di Airflow.

    • Se l'esecuzione del DAG non è riuscita, prova a risolvere i problemi del DAG finché non viene eseguita correttamente in Airflow 2.

      Se necessario, puoi sempre eseguire il fallback alla versione 1.10.* di Airflow del DAG:

      1. Metti in pausa il DAG nell'ambiente Airflow 2.

      2. Riattiva la messa in pausa del DAG nell'ambiente Airflow 1.10.*. Viene pianificata una nuova esecuzione del DAG per la stessa data e ora dell'esecuzione del DAG non riuscita.

      3. Quando è tutto pronto per continuare con la versione Airflow 2 del DAG, modifica la data di inizio, carica la nuova versione del DAG nell'ambiente Airflow 2 e ripeti la procedura.

Passaggio 9: monitora l'ambiente Airflow 2

Dopo aver trasferito tutti i DAG e la configurazione nell'ambiente Airflow 2, monitoralo per rilevare potenziali problemi, esecuzioni di DAG non riuscite e lo stato generale dell'ambiente. Se l'ambiente Airflow 2 funziona senza problemi per un periodo di tempo sufficiente, puoi rimuovere l'ambiente Airflow 1.10.*.

Passaggi successivi