Administrar conexiones de Airflow

Cloud Composer 1 | Cloud Composer 2

En esta página, se describe cómo administrar las conexiones de Airflow en tu entorno y acceder a ellas desde tus DAG.

Acerca de las conexiones de Airflow

Las conexiones de AIflow almacenan credenciales y otra información de conexión, como nombres de usuario, strings de conexión y contraseñas. Los DAG usan conexiones para comunicarse y acceder a los recursos en Google Cloud y otros servicios desde los DAG.

Los operadores de Airflow en tus DAG usan una conexión predeterminada para el operador o especificas un nombre de conexión personalizado.

Información acerca de la seguridad de la conexión

La mayoría de los operadores de Airflow no aceptan credenciales directamente. En su lugar, usan conexiones de Airflow.

Cuando creas un entorno nuevo, Cloud Composer genera una clave fernet única y permanente para el entorno y protege los servicios adicionales de conexión de forma predeterminada. Puedes ver fernet_key en la página Configuración de la IU de Airflow.

Para obtener más información sobre cómo se protegen las conexiones y contraseñas en Airflow, consulta Protege las conexiones y Enmascara datos sensibles.

Información acerca de los tipos de conexión

Airflow usa conexiones de diferentes tipos para conectarse a servicios específicos. Por ejemplo, el tipo de conexión de Google Cloud se conecta a otros servicios en Google Cloud. Como otro ejemplo, el tipo de conexión S3 se conecta a un bucket de Amazon S3.

Para agregar un tipo de conexión a Airflow, instala un paquete de PyPI con ese tipo de conexión. Algunos paquetes están preinstalados en tu entorno. Por ejemplo, puedes usar la conexión desde el paquete apache-airflow-providers-google sin instalar paquetes personalizados de PyPI.

Conexiones preconfiguradas

Cloud Composer configura las siguientes conexiones predeterminadas en tu entorno. Puedes usar estas conexiones para acceder a los recursos de tu proyecto sin configurarlas.

  • google_cloud_default
  • bigquery_default
  • google_cloud_datastore_default
  • google_cloud_storage_default

Agrega una conexión en Secret Manager

Puedes almacenar una conexión en Secret Manager sin agregarla a Airflow. Recomendamos usar este enfoque cuando almacenes credenciales y otra información sensible.

Para agregar una conexión en Secret Manager, haz lo siguiente:

  1. Configura Secret Manager para tu entorno.

  2. Agrega un secreto con el nombre que coincida con el patrón para las conexiones.

    Por ejemplo: airflow-connections-example_connection. En tus DAG, usa el nombre de la conexión sin el prefijo: example_connection.

  3. Agrega parámetros para la conexión:

    Formato JSON

    Agrega la representación JSON de tu conexión como el valor del secreto. Por ejemplo:

    {
  "conn_type": "mysql",
  "host": "example.com",
  "login": "login",
  "password": "password",
  "port": "9000"
}

    Para obtener más información sobre el formato de conexión JSON, consulta la documentación de Airflow.

    Formato URI

    Agrega la representación de URI de tu conexión como el valor del secreto:

    • El secreto debe almacenar una representación URI de la conexión. Por ejemplo, mysql://login:password@example.com:9000

    • El URI debe estar codificado como URL. Por ejemplo, una contraseña que tenga un símbolo de espacio debe codificarse como URL de la siguiente manera: mysql://login:secret%20password@example.com:9000.

    Airflow tiene un método útil para generar el URI de conexión. Puedes encontrar un ejemplo de cómo codificar una URL compleja con extras JSON en la documentación de Airflow.

  4. Verifica que todos los parámetros de conexión se leen de forma correcta desde Secret Manager.

Agregar una conexión en Airflow

Como alternativa al almacenamiento de tus conexiones en Secret Manager, puedes almacenarlas en Airflow.

Para agregar una conexión en Airflow, haz lo siguiente:

CLI de Airflow

Ejecuta el comando connections add de la CLI de Airflow mediante Google Cloud CLI. Por ejemplo:

En Airflow 2:

gcloud composer environments run ENVIRONMENT_NAME \
  --location LOCATION \
  connections add -- \
    --conn-type "mysql" \
    --conn-host "example.com" \
    --conn-port "9000" \
    --conn-login "login" \
    --conn-password "password" \
    example_connection

También puedes usar el argumento --conn-uri:

gcloud composer environments run ENVIRONMENT_NAME \
  --location LOCATION \
  connections add -- \
    --conn-uri "mysql://login:password@example.com:9000" \
    example_connection

En Airflow 1:

gcloud composer environments run ENVIRONMENT_NAME \
  --location LOCATION \
  connections -- \
    --add \
    --conn_type "mysql" \
    --conn_host "example.com" \
    --conn_port "9000" \
    --conn_login "login" \
    --conn_password "password" \
    --conn_id "example_connection"

Reemplaza lo siguiente:

  • ENVIRONMENT_NAME: Es el nombre de tu entorno.
  • LOCATION: Es la región en la que se encuentra el entorno.

IU de Airflow

Sigue la documentación de Airflow sobre la creación de conexiones.

Comprueba que Airflow lea correctamente una conexión

Puedes ejecutar el comando connections get de la CLI de Airflow mediante Google Cloud CLI para verificar que una conexión se lea de forma correcta. Por ejemplo, si almacenas una conexión en Secret Manager, esto proporciona una forma de verificar si Airflow lee todos los parámetros de una conexión desde un secreto.

gcloud composer environments run ENVIRONMENT_NAME \
    --location LOCATION \
    connections get \
    -- CONNECTION_NAME

Reemplaza lo siguiente:

  • ENVIRONMENT_NAME por el nombre del entorno.
  • LOCATION por la región en la que se encuentra el entorno
  • CONNECTION_NAME por el nombre de la conexión. Si tu conexión se almacena en Secret Manager, usa el nombre de la conexión sin el prefijo de conexión. Por ejemplo, especifica example_connection en lugar de airflow-connections-example_connection_json.

Ejemplo:

gcloud composer environments run example-environment \
    --location us-central1 \
    connections get \
    -- example_connection -o json

Usa conexiones de Airflow en tus DAG

En esta sección, se muestra cómo acceder a tu conexión desde un DAG.

Usa una conexión de Secret Manager

Usa el nombre de la conexión sin el prefijo. Por ejemplo, si tu secreto se llama airflow-connections-aws_s3, especifica aws_s3.

transfer_dir_from_s3 = S3ToGCSOperator(
    task_id='transfer_dir_from_s3',
    aws_conn_id='aws_s3',
    prefix='data-for-gcs',
    bucket='example-s3-bucket-transfer-operators',
    dest_gcs='gs://us-central1-example-environ-361f4221-bucket/data/from-s3/')

Si almacenas una conexión predeterminada en Secret Manager, puedes omitir el nombre de la conexión. Consulta la documentación de Airflow para un operador específico a fin de obtener el nombre de conexión predeterminado que usa un operador. Por ejemplo, el operador S3ToGCSOperator de Airflow usa la conexión aws_default de forma predeterminada. Puedes almacenar esta conexión predeterminada en un secreto llamado airflow-connections-aws_default.

Usar una conexión almacenada en Airflow

Usa el nombre de la conexión, como se define en Airflow:

transfer_dir_from_s3 = S3ToGCSOperator(
    task_id='transfer_dir_from_s3',
    aws_conn_id='aws_s3',
    prefix='data-for-gcs',
    bucket='example-s3-bucket-transfer-operators',
    dest_gcs='gs://us-central1-example-environ-361f4221-bucket/data/from-s3/')

Si deseas usar la conexión predeterminada para un operador, omite el nombre de la conexión. Consulta la documentación de Airflow para un operador específico a fin de obtener el nombre de conexión predeterminado que usa un operador. Por ejemplo, el operador S3ToGCSOperator de Airflow usa la conexión aws_default de forma predeterminada.

Soluciona problemas

Si tu entorno no puede acceder al secreto almacenado en Secret Manager, haz lo siguiente:

  1. Asegúrate de que Secret Manager esté configurado en tu entorno.

  2. Verifica que el nombre de la conexión en Secret Manager corresponda a la conexión que usa Airflow. Por ejemplo, para una conexión llamada example_connection, el nombre del secreto es airflow-connections-example_connection.

  3. Comprueba que Airflow lee de forma correcta una conexión.

¿Qué sigue?