O Cloud Composer 1 está no modo pós-manutenção. O Google não lança mais atualizações para o Cloud Composer 1, incluindo novas versões do Airflow, correções de bugs e atualizações de segurança. Recomendamos planejar a migração para o Cloud Composer 2.

Esta página foi traduzida pela API Cloud Translation.

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:

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

Acessar "Ambientes"
Na lista de ambientes, clique no nome do seu ambiente. A página Detalhes do ambiente é aberta.
Acesse a guia Registros.
Acesse Todos os registros > Registros do Airflow > Workers.
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.