Cloud Composer 1 se encuentra en modo posterior al mantenimiento. Google no lanza más actualizaciones para Cloud Composer 1, incluidas las versiones nuevas de Airflow, correcciones de errores y actualizaciones de seguridad. Recomendamos planificar la migración a Cloud Composer 2.

Se usó la API de Cloud Translation para traducir esta página.

Usa CeleryKubernetesExecutor

Cloud Composer 1 | Cloud Composer 2 | Cloud Composer 3

En esta página, se explica cómo habilitar CeleryKubernetesExecutor en Cloud Composer y cómo usar KubernetesExecutor en tus DAG.

Acerca de CeleryKubernetesExecutor

CeleryKubernetesExecutor es un tipo de ejecutor que puede usar CeleryExecutor y KubernetesExecutor al mismo tiempo. Airflow selecciona el ejecutor en función de la cola que definas para la tarea. En un DAG, puedes ejecutar algunas tareas con CeleryExecutor y otras con KubernetesExecutor:

CeleryExecutor está optimizado para ejecutar tareas de forma rápida y escalable.
KubernetesExecutor está diseñado para ejecutar tareas que consumen muchos recursos y ejecutar tareas de forma aislada.

CeleryKubernetesExecutor en Cloud Composer

CeleryKubernetesExecutor en Cloud Composer ofrece la capacidad de usar KubernetesExecutor para tus tareas. No es posible usar KubernetesExecutor en Cloud Composer por separado de CeleryKubernetesExecutor.

Cloud Composer ejecuta las tareas que ejecutas con KubernetesExecutor en el clúster de tu entorno, en el mismo espacio de nombres con los trabajadores de Airflow. Estas tareas tienen las mismas vinculaciones que los trabajadores de Airflow y pueden acceder a los recursos de tu proyecto.

Las tareas que ejecutas con KubernetesExecutor usan el modelo de precios de Cloud Composer, ya que los Pods con estas tareas se ejecutan en el clúster de tu entorno. Los SKU de Compute de Cloud Composer (para CPU, memoria y almacenamiento) se aplican a estos Pods.

Recomendamos ejecutar tareas con CeleryExecutor en los siguientes casos:

El tiempo de inicio de la tarea es importante.
Las tareas no requieren aislamiento del entorno de ejecución y no consumen muchos recursos.

Recomendamos ejecutar tareas con KubernetesExecutor en los siguientes casos:

Las tareas requieren aislamiento del entorno de ejecución. Por ejemplo, para que las tareas no compitan por la memoria y la CPU, ya que se ejecutan en sus propios Pods.
Las tareas requieren bibliotecas de sistema adicionales (o paquetes de PyPI).
Las tareas consumen muchos recursos y quieres controlar los recursos de CPU y memoria disponibles.

Comparación entre KubernetesExecutor y KubernetesPodOperator

Ejecutar tareas con KubernetesExecutor es similar a ejecutar tareas con KubernetesPodOperator. Las tareas se ejecutan en pods, lo que proporciona aislamiento de tareas a nivel de pod y una mejor administración de recursos.

Sin embargo, existen algunas diferencias clave:

KubernetesExecutor ejecuta tareas solo en el espacio de nombres de Cloud Composer con versión de tu entorno. No es posible cambiar este espacio de nombres en Cloud Composer. Puedes especificar un espacio de nombres en el que KubernetesPodOperator ejecuta las tareas del Pod.
KubernetesExecutor puede usar cualquier operador de Airflow integrado. KubernetesPodOperator solo ejecuta una secuencia de comandos proporcionada que define el punto de entrada del contenedor.
KubernetesExecutor usa la imagen predeterminada de Docker de Cloud Composer con el mismo Python, las mismas anulaciones de opciones de configuración de Airflow, las variables de entorno y los paquetes de PyPI que se definen en tu entorno de Cloud Composer.

Acerca de las imágenes de Docker

De forma predeterminada, KubernetesExecutor inicia tareas con la misma imagen de Docker que usa Cloud Composer para los trabajadores de Celery. Esta es la imagen de Cloud Composer de tu entorno, con todos los cambios que especificaste para este, como los paquetes de PyPI personalizados o las variables de entorno.

Antes de comenzar

Puedes usar CeleryKubernetesExecutor en Cloud Composer 3.
No es posible usar ningún ejecutor que no sea CeleryKubernetesExecutor en Cloud Composer 3. Esto significa que puedes ejecutar tareas con CeleryExecutor, KubernetesExecutor o ambos en un DAG, pero no es posible configurar tu entorno para que solo use KubernetesExecutor o CeleryExecutor.

Configura CeleryKubernetesExecutor

Es posible que quieras anular las opciones de configuración existentes de Airflow que se relacionan con KubernetesExecutor:

[kubernetes]worker_pods_creation_batch_size

Esta opción define la cantidad de llamadas de creación de Pods de trabajadores de Kubernetes por bucle de programador. El valor predeterminado es 1, por lo que solo se inicia un Pod por señal de monitoreo de funcionamiento del programador. Si usas mucho KubernetesExecutor, recomendamos aumentar este valor.
[kubernetes]worker_pods_pending_timeout

Esta opción define, en segundos, cuánto tiempo puede permanecer un trabajador en el estado Pending (se está creando el Pod) antes de que se considere con errores. El valor predeterminado es 5 minutos.

Ejecuta tareas con KubernetesExecutor o CeleryExecutor

Puedes ejecutar tareas con CeleryExecutor, KubernetesExecutor o ambos en un DAG:

Para ejecutar una tarea con KubernetesExecutor, especifica el valor kubernetes en el parámetro queue de una tarea.
Para ejecutar una tarea con CeleryExecutor, omite el parámetro queue.

En el siguiente ejemplo, se ejecuta la tarea task-kubernetes con KubernetesExecutor y la tarea task-celery con CeleryExecutor:

import datetime
import airflow
from airflow.operators.python_operator import PythonOperator

with airflow.DAG(
  "composer_sample_celery_kubernetes",
  start_date=datetime.datetime(2022, 1, 1),
  schedule_interval="@daily") as dag:

  def kubernetes_example():
      print("This task runs using KubernetesExecutor")

  def celery_example():
      print("This task runs using CeleryExecutor")

  # To run with KubernetesExecutor, set queue to kubernetes
  task_kubernetes = PythonOperator(
    task_id='task-kubernetes',
    python_callable=kubernetes_example,
    dag=dag,
    queue='kubernetes')

  # To run with CeleryExecutor, omit the queue argument
  task_celery = PythonOperator(
    task_id='task-celery',
    python_callable=celery_example,
    dag=dag)

  task_kubernetes >> task_celery

Ejecuta comandos de la CLI de Airflow relacionados con KubernetesExecutor

Puedes ejecutar varios comandos de la CLI de Airflow relacionados con KubernetesExecutor mediante gcloud.

Personaliza las especificaciones del Pod de trabajador

Para personalizar las especificaciones del pod de trabajador, puedes pasarlo en el parámetro executor_config de una tarea. Puedes usar esto para definir requisitos personalizados de CPU y memoria.

Puedes anular toda la especificación del Pod de trabajador que se usa para ejecutar una tarea. Para recuperar la especificación de Pod de una tarea que usa KubernetesExecutor, puedes ejecutar el comando kubernetes generate-dag-yaml de la CLI de Airflow.

Para obtener más información sobre cómo personalizar las especificaciones del pod de trabajador, consulta la documentación de Airflow.

En el siguiente ejemplo, se muestra una tarea que usa una especificación de pod de trabajador personalizado:

PythonOperator(
    task_id='custom-spec-example',
    python_callable=f,
    dag=dag,
    queue='kubernetes',
    executor_config={
        'pod_override': k8s.V1Pod(
            spec=k8s.V1PodSpec(
                containers=[
                    k8s.V1Container(
                        name='base',
                        resources=k8s.V1ResourceRequirements(requests={
                            'cpu': '500m',
                            'memory': '1000Mi',
                        })
                    ),
                ],
            ),
        )
    },
)

Ver registros de tareas

Los registros de tareas que ejecuta KubernetesExecutor están disponibles en la pestaña Registros, junto con los registros de tareas que ejecuta CeleryExecutor:

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

Ir a Entornos
En la lista de entornos, haz clic en el nombre de tu entorno. Se abrirá la página Detalles del entorno.
Ve a la pestaña Registros.
Navega a Todos los registros > Registros de Airflow > Trabajadores.
Los trabajadores llamados airflow-k8s-worker ejecutan tareas de KubernetesExecutor. Para buscar registros de una tarea específica, puedes usar un ID de DAG o un ID de tarea como palabra clave en la búsqueda.