Cloud Composer 1 está en el modo posterior al mantenimiento. Google no lanzará más actualizaciones de Cloud Composer 1, incluidas nuevas versiones 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 de ejecución 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 la ejecución rápida y escalable de tareas.
KubernetesExecutor está diseñado para ejecutar tareas que consumen muchos recursos y ejecutar tareas aisladas.

CeleryKubernetesExecutor en Cloud Composer

CeleryKubernetesExecutor en Cloud Composer te permite usar KubernetesExecutor para tus tareas. No es posible utilizar 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 estos las tareas que 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.

Te recomendamos que ejecutes 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.

Te recomendamos que ejecutes 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 las tareas recursos de CPU y memoria.

KubernetesExecutor en comparación con 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 del 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 ejecute tareas de pod.
KubernetesExecutor puede usar cualquier operador de Airflow integrado. KubernetesPodOperator ejecuta solo una secuencia de comandos proporcionada que define el punto de entrada del contenedor.
KubernetesExecutor usa la imagen predeterminada de Docker de Cloud Composer con las mismas anulaciones de opciones de configuración de Python y 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 para tu entorno, con todos los cambios que especificaste para tu entorno, como paquetes de PyPI personalizados o 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 la configuración existente de Airflow. relacionadas 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 del programador. El valor predeterminado es 1, por lo que solo se inicia un solo pod por cada mensaje de estado del programador. Si usas KubernetesExecutor con frecuencia, te recomendamos que aumentes 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 que falló. 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 con gcloud.

Personaliza la especificación del pod de trabajo

Para personalizar las especificaciones del Pod del trabajador, pásalo al archivo executor_config. parámetro de una tarea. Puedes usarlo para definir parámetros de configuración y los requisitos de cumplimiento.

Puedes anular toda la especificación del Pod de trabajador que se usa para ejecutar una tarea. Para recuperar la especificación del 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 trabajo, consulta la documentación de Airflow.

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

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',
                        })
                    ),
                ],
            ),
        )
    },
)

Cómo ver los 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 de una tarea como palabra clave en la búsqueda.