Usar o CeleryKubernetesExecutor

Cloud Composer 1 | Cloud Composer 2 | Cloud Composer 3

Nesta página, explicamos como ativar o CeleryKubernetesExecutor no Cloud Composer e como usar KubernetesExecutor nos DAGs.

Sobre o CeleryKubernetesExecutor

CeleryKubernetesExecutor é um tipo de executor que pode usar CeleryExecutor e KubernetesExecutor ao mesmo tempo. O Airflow seleciona o executor com base na fila que você define para a tarefa. Em um DAG, é possível executar algumas tarefas com o CeleryExecutor e outras com o KubernetesExecutor:

  • O CeleryExecutor é otimizado para execução rápida e escalonável de tarefas.
  • O KubernetesExecutor foi projetado para executar tarefas que consomem muitos recursos e executar tarefas isoladamente.

CeleryKubernetesExecutor no Cloud Composer

O CeleryKubernetesExecutor no Cloud Composer permite o uso do KubernetesExecutor para suas tarefas. Não é possível usar KubernetesExecutor no Cloud Composer separado do o CeleryKubernetesExecutor.

O Cloud Composer executa tarefas que você executa com KubernetesExecutor no cluster do ambiente e no mesmo namespace que os workers do Airflow. Essas as tarefas têm as mesmas vinculações que o Airflow workers e pode acessar recursos no seu projeto.

As tarefas que você executa com KubernetesExecutor usam o Modelo de preços do Cloud Composer, já que os pods com essas as tarefas são executadas no cluster do ambiente. As SKUs de computação do Cloud Composer (para CPU, memória e armazenamento) se aplicam a esses pods.

Recomendamos executar tarefas com o CeleryExecutor quando:

  • O tempo de inicialização da tarefa é importante.
  • As tarefas não exigem isolamento de execução e não consomem muitos recursos.

Recomendamos executar tarefas com o KubernetesExecutor quando:

  • As tarefas exigem isolamento do ambiente de execução. Por exemplo, para que as tarefas não concorram memória e CPU, já que são executados nos próprios pods.
  • As tarefas exigem bibliotecas de sistema adicionais (ou pacotes PyPI).
  • As tarefas consomem muitos recursos, e você quer controlar os recursos de CPU e memória disponíveis.

KubernetesExecutor em comparação com KubernetesPodOperator

A execução de tarefas com KubernetesExecutor é semelhante à executar tarefas usando o KubernetesPodOperator. As tarefas são executadas em os pods, fornecendo isolamento de tarefas no nível do pod e melhor gerenciamento de recursos.

No entanto, há algumas diferenças importantes:

  • KubernetesExecutor executa tarefas apenas no Cloud Composer com controle de versão namespace do seu ambiente. Não é possível mudar esse namespace no Cloud Composer. É possível especificar um namespace em que o KubernetesPodOperator executa tarefas de pod.
  • KubernetesExecutor pode usar qualquer operador integrado do Airflow. KubernetesPodOperator executa apenas um script fornecido definido pelo ponto de entrada do contêiner.
  • O KubernetesExecutor usa a imagem padrão do Docker do Cloud Composer com as mesmas substituições de opção de configuração do Python e do Airflow, variáveis de ambiente e pacotes PyPI definidos no ambiente do Cloud Composer.

Sobre as imagens do Docker

Por padrão, o KubernetesExecutor inicia tarefas usando a mesma imagem do Docker que o Cloud Composer usa para workers do Celery. Esta é a imagem do Cloud Composer para seu ambiente, com todas as mudanças especificadas, como pacotes PyPI personalizados ou variáveis de ambiente.

Antes de começar

  • É possível usar o CeleryKubernetesExecutor no Cloud Composer 3.

  • Não é possível usar nenhum executor, exceto o CeleryKubernetesExecutor, no Cloud Composer 3. Isso significa que você pode executar tarefas usando CeleryExecutor, KubernetesExecutor ou ambos em um DAG, mas não possível configurar seu ambiente para usar apenas KubernetesExecutor ou o CeleryExecutor.

Configurar o CeleryKubernetesExecutor

substitua a configuração atual do Airflow opções relacionadas a KubernetesExecutor:

  • [kubernetes]worker_pods_creation_batch_size

    Essa opção define o número de chamadas de criação de pods de worker do Kubernetes por ciclo de agendamento. O valor padrão é 1, então apenas um pod é iniciado por sinal de funcionamento do programador. Se você usa o KubernetesExecutor com frequência, recomendamos aumentar esse valor.

  • [kubernetes]worker_pods_pending_timeout

    Essa opção define, em segundos, por quanto tempo um worker pode permanecer no estado Pending (o pod está sendo criado) antes de ser considerado com falha. O valor padrão é 5 minutos.

Executar tarefas com KubernetesExecutor ou CeleryExecutor

É possível executar tarefas usando o CeleryExecutor, KubernetesExecutor ou ambos em um DAG:

  • Para executar uma tarefa com o KubernetesExecutor, especifique o valor kubernetes no parâmetro queue de uma tarefa.
  • Para executar uma tarefa com o CeleryExecutor, omita o parâmetro queue.

O exemplo a seguir executa a tarefa task-kubernetes usando o KubernetesExecutor e a tarefa task-celery usando o 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

Executar comandos da CLI do Airflow relacionados ao KubernetesExecutor

É possível executar vários comandos da CLI do Airflow relacionados ao KubernetesExecutor usando gcloud.

Personalizar a especificação do pod de worker

É possível personalizar a especificação do pod de worker transmitindo-a no parâmetro executor_config de uma tarefa. Você pode usar isso para definir requisitos personalizados de CPU e memória.

É possível substituir toda a especificação do pod de worker usada para executar uma tarefa. Para extrair a especificação do pod de uma tarefa usada pelo KubernetesExecutor, execute o comando CLI kubernetes generate-dag-yaml do Airflow.

Para mais informações sobre como personalizar a especificação do pod de worker, consulte a documentação do Airflow.

O exemplo a seguir demonstra uma tarefa que usa a especificação do pod de worker 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 tarefas

Os registros de tarefas executadas pelo KubernetesExecutor estão disponíveis na guia Logs, junto com os registros de tarefas executadas pelo CeleryExecutor:

  1. No console do Google Cloud, acesse a página Ambientes.

    Acessar "Ambientes"

  2. Na lista de ambientes, clique no nome do seu ambiente. A página Detalhes do ambiente é aberta.

  3. Acesse a guia Registros.

  4. Acesse Todos os registros > Registros do Airflow > Workers.

  5. Execução de workers chamados airflow-k8s-worker KubernetesExecutor. Para procurar os registros de uma tarefa específica, usar um ID do DAG ou da tarefa como uma palavra-chave na pesquisa.

A seguir