Migrar entornos a Airflow 2

Cloud Composer 3 | Cloud Composer 2 | Cloud Composer 1

En esta página se explica cómo transferir DAGs, datos y configuración de tus entornos de Airflow 1.10.* a entornos con Airflow 2 y versiones posteriores.

Otras guías de migración

De Para Método Guía
Cloud Composer 2 Cloud Composer 3 En paralelo, con la secuencia de comandos de migración Guía de migración de secuencias de comandos
Cloud Composer 2 Cloud Composer 3 En paralelo, con capturas Guía de migración de capturas
Cloud Composer 1, Airflow 2 Cloud Composer 3 En paralelo, con capturas Guía de migración de capturas
Cloud Composer 1, Airflow 2 Cloud Composer 2 En paralelo, con capturas Guía de migración de capturas
Cloud Composer 1, Airflow 2 Cloud Composer 2 Transferencia manual en paralelo Guía de migración manual
Cloud Composer 1, Airflow 1 Cloud Composer 2, Airflow 2 En paralelo, con capturas Guía de migración de capturas
Cloud Composer 1, Airflow 1 Cloud Composer 2, Airflow 2 Transferencia manual en paralelo Guía de migración manual
Cloud Composer 1, Airflow 1 Cloud Composer 1, Airflow 2 Transferencia manual en paralelo Esta guía

Actualizaciones en paralelo

Cloud Composer proporciona la secuencia de comandos de transferencia de bases de datos de Cloud Composer para migrar la base de datos de metadatos, los DAGs, los datos y los complementos de los entornos de Cloud Composer con Airflow 1.10.14 y Airflow 1.10.15 a los entornos de Cloud Composer con Airflow 2.0.1 y versiones posteriores de Airflow.

Esta es una forma alternativa a la que se describe en esta guía. Algunas partes de esta guía siguen siendo válidas cuando se usa la secuencia de comandos proporcionada. Por ejemplo, puede que quieras comprobar si tus DAGs son compatibles con Airflow 2 antes de migrarlos o asegurarte de que no se ejecuten DAGs simultáneamente y de que no haya ejecuciones de DAGs adicionales o que falten.

Antes de empezar

Antes de empezar a usar entornos de Cloud Composer con Airflow 2, ten en cuenta los cambios que Airflow 2 introduce en los entornos de Cloud Composer.

Varias programaciones

Puedes usar más de un programador de Airflow en tu entorno. Puedes definir el número de programadores cuando crees un entorno o actualices uno que ya tengas.

Ejecutor de Celery y Kubernetes

Airflow 2 Celery+Kubernetes Executor se admite en Cloud Composer 3.

Puntos de ruptura

Airflow 2 introduce muchos cambios importantes, algunos de los cuales son incompatibles con versiones anteriores:

Diferencias entre los entornos con Airflow 2 y Airflow 1.10.*

Principales diferencias entre los entornos de Cloud Composer con Airflow 1.10.* y los entornos con Airflow 2:

  • Los entornos con Airflow 2 usan Python 3.8. Es una versión más reciente que la que se usa en los entornos de Airflow 1.10.*. No se admiten Python 2, Python 3.6 ni Python 3.7.
  • Airflow 2 usa un formato de CLI diferente. Cloud Composer admite el nuevo formato en entornos con Airflow 2 mediante el comando gcloud composer environments run.
  • Los paquetes PyPI preinstalados son diferentes en los entornos de Airflow 2. Para ver una lista de los paquetes PyPI preinstalados, consulta la lista de versiones de Cloud Composer.
  • La serialización de DAGs siempre está habilitada en Airflow 2. Por lo tanto, ya no es necesario cargar DAGs de forma asíncrona, y esta función no es compatible con Airflow 2. Por este motivo, no se admite la configuración de los parámetros [core]store_serialized_dags y [core]store_dag_code en Airflow 2, y los intentos de configurarlos se notificarán como errores.
  • No se admiten los complementos del servidor web de Airflow. Esto no afecta a los complementos de programador o de trabajador, incluidos los operadores y sensores de Airflow.
  • En los entornos de Airflow 2, el rol de usuario predeterminado de Airflow es Op. En los entornos con Airflow 1.10.*, el rol predeterminado es Admin.

Paso 1: Comprueba la compatibilidad con Airflow 2

Para comprobar si hay posibles conflictos con Airflow 2, consulta la guía de actualización a Airflow 2.0+ en la sección sobre actualización de DAGs.

Un problema habitual que puede surgir está relacionado con las rutas de importación incompatibles. Para obtener más información sobre cómo solucionar este problema de compatibilidad, consulta la sección sobre proveedores de versiones anteriores de la guía de actualización a Airflow 2.0 o versiones posteriores.

Paso 2: Crea un entorno de Airflow 2, transfiere las anulaciones de configuración y las variables de entorno

Crea un entorno de Airflow 2 y transfiere las anulaciones de configuración y las variables de entorno:

  1. Sigue los pasos para crear un entorno. Antes de crear un entorno, también debe especificar las anulaciones de configuración y las variables de entorno, como se explica más adelante.

  2. Cuando selecciones una imagen, elige una imagen con Airflow 2.

  3. Transfiere manualmente los parámetros de configuración de tu entorno de Airflow 1.10.* al nuevo entorno de Airflow 2.

    Consola

    1. Cuando cree un entorno, despliegue la sección Redes, anulaciones de la configuración de Airflow y funciones adicionales.

    2. En Anulaciones de configuración de Airflow, haz clic en Añadir anulación de configuración de Airflow.

    3. Copia todas las anulaciones de configuración de tu entorno de Airflow 1.10.*.

      Algunas opciones de configuración usan un nombre y una sección diferentes en Airflow 2. Para obtener más información, consulta Cambios en la configuración.

    4. En Variables de entorno, haz clic en Añadir variable de entorno.

    5. Copia todas las variables de entorno de tu entorno de Airflow 1.10.*.

    6. Haz clic en Crear para crear un entorno.

Paso 3: Instala paquetes de PyPI en el entorno de Airflow 2

Una vez que se haya creado tu entorno de Airflow 2, instala los paquetes de PyPI en él:

Consola

  1. En la Google Cloud consola, ve a la página Entornos.

    Ir a Entornos

  2. Selecciona tu entorno de Airflow 2.

  3. Ve a la pestaña Paquetes de PyPI y haz clic en Editar.

  4. Copia los requisitos del paquete PyPI de tu entorno de Airflow 1.10.*. Haz clic en Guardar y espera a que se actualice el entorno.

    Como los entornos de Airflow 2 usan un conjunto diferente de paquetes preinstalados y una versión de Python distinta, es posible que te encuentres con conflictos de paquetes PyPI difíciles de resolver.

Paso 4: Transfiere variables y grupos a Airflow 2

Airflow 1.10.* permite exportar variables y grupos a archivos JSON. Después, puedes importar estos archivos a tu entorno de Airflow 2.

Solo tienes que transferir los grupos si tienes grupos personalizados que no sean default_pool. De lo contrario, omite los comandos que exportan e importan grupos.

gcloud

  1. Exporta variables de tu entorno de Airflow 1.10.*:

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

    Sustituye:

    • AIRFLOW_1_ENV con el nombre de tu entorno de Airflow 1.10.*.
    • AIRFLOW_1_LOCATION con la región en la que se encuentra el entorno.

  2. Exporta los pools de tu entorno de Airflow 1.10.*:

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

  3. Obtén el URI del segmento de tu entorno de Airflow 2.

    1. Ejecuta el siguiente comando:

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

      Sustituye:

      • AIRFLOW_2_ENV con el nombre de tu entorno de Airflow 2.
      • AIRFLOW_2_LOCATION por la región en la que se encuentra el entorno.

    2. En el resultado, elimina la carpeta /dags. El resultado es el URI del cubo de tu entorno de Airflow 2.

      Por ejemplo, cambia gs://us-central1-example-916807e1-bucket/dags por gs://us-central1-example-916807e1-bucket.

  4. Transfiere archivos JSON con variables y grupos a tu entorno de 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

    Sustituye AIRFLOW_2_BUCKET por el URI del bucket de tu entorno de Airflow 2, que has obtenido en el paso anterior.

  5. Importa variables y grupos a 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. Comprueba que se hayan importado las variables y los grupos:

    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. Elimina los archivos JSON de los contenedores:

    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

Paso 5: Transfiere otros datos del segmento de tu entorno Airflow 1.10.*

gcloud

  1. Transfiere plugins a tu entorno de Airflow 2. Para ello, exporta los complementos del bucket de tu entorno de Airflow 1.10.* a la carpeta /plugins del bucket de tu entorno de Airflow 2:

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

  2. Comprueba que la carpeta /plugins se ha importado correctamente:

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

  3. Exporta la carpeta /data de tu entorno de Airflow 1.10.* al entorno de Airflow 2:

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

  4. Comprueba que la carpeta /data se ha importado correctamente:

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

Paso 6: Transfiere conexiones y usuarios

Airflow 1.10.* no admite la exportación de usuarios ni de conexiones. Para transferir usuarios y conexiones, crea manualmente nuevas cuentas de usuario y conexiones en tu entorno de Airflow 2.

gcloud

  1. Para obtener una lista de las conexiones de tu entorno de Airflow 1.10.*, ejecuta el siguiente comando:

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

  2. Para crear una conexión en tu entorno de Airflow 2, ejecuta el connectionscomando de la CLI de Airflow a través de gcloud. Por ejemplo:

    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. Para ver una lista de los usuarios de tu entorno de Airflow 1.10.*, sigue estos pasos:

    1. Abre la interfaz web de Airflow de tu entorno de Airflow 1.10.*.

    2. Vaya a Administrar > Usuarios.

  4. Para crear una cuenta de usuario en tu entorno de Airflow 2, ejecuta el users createcomando de la CLI de Airflow a través de gcloud. Por ejemplo:

    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

Paso 7: Asegúrate de que tus DAGs estén listos para Airflow 2

Antes de transferir los DAGs a tu entorno de Airflow 2, asegúrate de lo siguiente:

  1. Tus DAGs se ejecutan correctamente y no hay problemas de compatibilidad.

  2. Tus DAGs usan instrucciones de importación correctas.

    Por ejemplo, la nueva instrucción de importación de BigQueryCreateDataTransferOperator puede tener el siguiente aspecto:

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

  3. Tus DAGs se han actualizado para Airflow 2. Este cambio es compatible con Airflow 1.10.14 y versiones posteriores.

Paso 8: Transfiere los DAGs al entorno de Airflow 2

Pueden surgir los siguientes problemas al transferir DAGs entre entornos:

  • Si un DAG está habilitado (no en pausa) en ambos entornos, cada entorno ejecuta su propia copia del DAG según la programación. Esto puede provocar que se ejecuten DAGs simultáneos con los mismos datos y tiempo de ejecución.

  • Debido a la puesta al día de DAGs, Airflow programa ejecuciones de DAGs adicionales, empezando por la fecha de inicio especificada en tus DAGs. Esto ocurre porque la nueva instancia de Airflow no tiene en cuenta el historial de ejecuciones de DAG del entorno 1.10.*. Esto puede provocar que se programen un gran número de ejecuciones de DAG a partir de la fecha de inicio especificada.

Evitar ejecuciones simultáneas de DAGs

En tu entorno de Airflow 2, anula la opción de configuración de dags_are_paused_at_creation Airflow. Después de hacer este cambio, todos los DAGs nuevos se pausarán de forma predeterminada.

Sección Clave Valor
core dags_are_paused_at_creation True

Evitar ejecuciones de DAGs adicionales o que falten

Especifica una nueva fecha de inicio estática en los DAGs que transfieras a tu entorno de Airflow 2.

Para evitar que haya intervalos de fechas de ejecución que se solapen o que no se cubran, la primera ejecución del DAG debe producirse en el entorno de Airflow 2 en la siguiente ocurrencia del intervalo de programación. Para ello, define la nueva fecha de inicio en tu DAG para que sea anterior a la fecha de la última ejecución en el entorno de Airflow 1.10.*.

Por ejemplo, si tu DAG se ejecuta a las 15:00, las 17:00 y las 21:00 todos los días en el entorno de Airflow 1.10.*, la última ejecución del DAG se produjo a las 15:00 y tienes previsto transferir el DAG a las 15:15, la fecha de inicio del entorno de Airflow 2 puede ser hoy a las 14:45. Después de habilitar el DAG en el entorno de Airflow 2, Airflow programa una ejecución del DAG para las 17:00.

Por ejemplo, si tu DAG se ejecuta a las 00:00 todos los días en el entorno de Airflow 1.10.*, la última ejecución del DAG se produjo a las 00:00 del 26 de abril del 2021 y tienes previsto transferir el DAG a las 13:00 del 26 de abril del 2021, la fecha de inicio del entorno de Airflow 2 puede ser las 23:45 del 25 de abril del 2021. Después de habilitar el DAG en el entorno de Airflow 2, Airflow programa una ejecución del DAG para las 00:00 del 27 de abril del 2021.

Transfiere tus DAGs uno a uno al entorno de Airflow 2

Sigue este procedimiento para transferir cada DAG:

  1. Asegúrate de que la nueva fecha de inicio del DAG se haya definido tal como se describe en la sección anterior.

  2. Sube el DAG actualizado al entorno de Airflow 2. Este DAG está en pausa en el entorno de Airflow 2 debido a la anulación de la configuración, por lo que aún no se ha programado ninguna ejecución de DAG.

  3. En la interfaz web de Airflow, ve a DAGs y comprueba si hay errores de sintaxis de DAG.

  4. En el momento en que tengas previsto transferir el DAG:

    1. Pausa el DAG en tu entorno de Airflow 1.10.*.

    2. Reanuda el DAG en tu entorno de Airflow 2.

    3. Comprueba que la nueva ejecución del DAG se haya programado a la hora correcta.

    4. Espera a que se ejecute el DAG en el entorno de Airflow 2 y comprueba si se ha completado correctamente.

  5. En función de si la ejecución del DAG se ha completado correctamente:

    • Si la ejecución del DAG se completa correctamente, puedes continuar y usar el DAG desde tu entorno de Airflow 2. Finalmente, puedes eliminar la versión 1.10.* del DAG de Airflow.

    • Si la ejecución del DAG ha fallado, intenta solucionar los problemas del DAG hasta que se ejecute correctamente en Airflow 2.

      Si es necesario, siempre puedes volver a la versión 1.10.* de Airflow del DAG:

      1. Pausa el DAG en tu entorno de Airflow 2.

      2. Reanuda el DAG en tu entorno de Airflow 1.10.*. De esta forma, se programa una nueva ejecución del DAG para la misma fecha y hora que la ejecución del DAG que ha fallado.

      3. Cuando quieras continuar con la versión de Airflow 2 del DAG, ajusta la fecha de inicio, sube la nueva versión del DAG a tu entorno de Airflow 2 y repite el procedimiento.

Paso 9: Monitoriza tu entorno de Airflow 2

Una vez que hayas transferido todos los DAGs y la configuración al entorno de Airflow 2, monitorízalo para detectar posibles problemas, ejecuciones de DAGs fallidas y el estado general del entorno. Si el entorno de Airflow 2 funciona sin problemas durante un periodo suficiente, puedes eliminar el entorno de Airflow 1.10.*.

Siguientes pasos