Il 15 settembre 2026, tutte le versioni di Cloud Composer 1 e le versioni 2.0.x di Cloud Composer 2 raggiungeranno la fine del ciclo di vita pianificato. Non potrai utilizzare gli ambienti con queste versioni. Ti consigliamo di pianificare la migrazione a Cloud Composer 3. Le versioni 2.1.x e successive di Cloud Composer 2 sono ancora supportate e non sono interessate da questa modifica.

Questa pagina è stata tradotta dall'API Cloud Translation.

Utilizzare CeleryKubernetesExecutor

Cloud Composer 3 | Cloud Composer 2 | Cloud Composer 1

Questa pagina spiega come attivare CeleryKubernetesExecutor in Cloud Composer e come utilizzare KubernetesExecutor nei tuoi DAG.

Informazioni su CeleryKubernetesExecutor

CeleryKubernetesExecutor è un tipo di esecutore che può utilizzare CeleryExecutor e KubernetesExecutor contemporaneamente. Airflow seleziona l'executor in base alla coda definita per l'attività. In un DAG, puoi eseguire alcune attività con CeleryExecutor e altre attività con KubernetesExecutor:

CeleryExecutor è ottimizzato per l'esecuzione rapida e scalabile delle attività.
KubernetesExecutor è progettato per l'esecuzione di attività che richiedono molte risorse e per l'esecuzione di attività in isolamento.

CeleryKubernetesExecutor in Cloud Composer

CeleryKubernetesExecutor in Cloud Composer offre la possibilità di utilizzare KubernetesExecutor per le tue attività. Non è possibile utilizzare KubernetesExecutor in Cloud Composer separatamente da CeleryKubernetesExecutor.

Cloud Composer esegue le attività che esegui con KubernetesExecutor nel cluster del tuo ambiente, nello stesso spazio dei nomi dei worker Airflow. Queste attività hanno gli stessi binding dei worker Airflow e possono accedere alle risorse del tuo progetto.

Le attività che esegui con KubernetesExecutor utilizzano il modello di prezzo di Cloud Composer, poiché i pod con queste attività vengono eseguiti nel cluster del tuo ambiente. A questi pod si applicano gli SKU di Cloud Composer Compute (per CPU, memoria e spazio di archiviazione).

Ti consigliamo di eseguire le attività con CeleryExecutor quando:

Il tempo di avvio dell'attività è importante.
Le attività non richiedono l'isolamento in fase di runtime e non richiedono molte risorse.

Ti consigliamo di eseguire le attività con KubernetesExecutor quando:

Le attività richiedono l'isolamento del runtime. Ad esempio, in modo che le attività non competano per memoria e CPU, poiché vengono eseguite nei propri pod.
Le attività richiedono molte risorse e vuoi controllare le risorse di CPU e memoria disponibili.

KubernetesExecutor rispetto a KubernetesPodOperator

L'esecuzione di attività con KubernetesExecutor è simile all'esecuzione di attività utilizzando KubernetesPodOperator. Le attività vengono eseguite nei pod, fornendo così l'isolamento delle attività a livello di pod e una migliore gestione delle risorse.

Tuttavia, esistono alcune differenze fondamentali:

KubernetesExecutor esegue le attività solo nello spazio dei nomi Cloud Composer con controllo delle versioni del tuo ambiente. Non è possibile modificare questo spazio dei nomi in Cloud Composer. Puoi specificare uno spazio dei nomi in cui KubernetesPodOperator esegue le attività del pod.
KubernetesExecutor può utilizzare qualsiasi operatore Airflow integrato. KubernetesPodOperator esegue solo uno script fornito definito dal punto di ingresso del container.
KubernetesExecutor utilizza l'immagine Docker Cloud Composer predefinita con gli stessi override di Python, delle opzioni di configurazione di Airflow, delle variabili di ambiente e dei pacchetti PyPI definiti nel tuo ambiente Cloud Composer.

Informazioni sulle immagini Docker

Per impostazione predefinita, KubernetesExecutor avvia le attività utilizzando la stessa immagine Docker che Cloud Composer utilizza per i worker Celery. Questa è l'immagine di Cloud Composer per il tuo ambiente, con tutte le modifiche che hai specificato per l'ambiente, ad esempio pacchetti PyPI personalizzati o variabili di ambiente.

Prima di iniziare

Puoi utilizzare CeleryKubernetesExecutor in Cloud Composer 3.
Non è possibile utilizzare un executor diverso da CeleryKubernetesExecutor in Cloud Composer 3. Ciò significa che puoi eseguire attività utilizzando CeleryExecutor, KubernetesExecutor o entrambi in un DAG, ma non è possibile configurare l'ambiente in modo che utilizzi solo KubernetesExecutor o CeleryExecutor.

Configura CeleryKubernetesExecutor

Potresti voler ignorare le opzioni di configurazione Airflow esistenti correlate a KubernetesExecutor:

[kubernetes]worker_pods_creation_batch_size

Questa opzione definisce il numero di chiamate di creazione di pod di lavoro Kubernetes per ciclo dello scheduler. Il valore predefinito è 1, quindi viene avviato un solo pod per heartbeat dello scheduler. Se utilizzi KubernetesExecutor di frequente, ti consigliamo di aumentare questo valore.
[kubernetes]worker_pods_pending_timeout

Questa opzione definisce, in secondi, per quanto tempo un worker può rimanere nello stato Pending (il pod è in fase di creazione) prima di essere considerato non riuscito. Il valore predefinito è 5 minuti.

Esegui attività con KubernetesExecutor o CeleryExecutor

Puoi eseguire attività utilizzando CeleryExecutor, KubernetesExecutor o entrambi in un unico DAG:

Per eseguire un'attività con KubernetesExecutor, specifica il valore kubernetes nel parametro queue di un'attività.
Per eseguire un'attività con CeleryExecutor, ometti il parametro queue.

L'esempio seguente esegue l'attività task-kubernetes utilizzando KubernetesExecutor e l'attività task-celery utilizzando 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="@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

Esegui i comandi della CLI Airflow relativi a KubernetesExecutor

Puoi eseguire diversi comandi della CLI Airflow relativi a KubernetesExecutor utilizzando gcloud.

Personalizza la specifica del pod worker

Puoi personalizzare la specifica del pod worker passandola nel parametro executor_config di un'attività. Puoi utilizzarlo per definire requisiti personalizzati di CPU e memoria.

Puoi eseguire l'override dell'intera specifica del pod worker utilizzata per eseguire un'attività. Per recuperare la specifica del pod di un'attività utilizzata da KubernetesExecutor, puoi eseguire il comando kubernetes generate-dag-yaml della CLI Airflow.

Per saperne di più sulla personalizzazione della specifica del pod worker, consulta la documentazione di Airflow.

Cloud Composer 3 supporta i seguenti valori per i requisiti delle risorse:

Risorsa	Minimo	Massimo	Passaggio
CPU	0,25	32	Valori del passo: 0,25, 0,5, 1, 2, 4, 6, 8, 10, ..., 32. I valori richiesti vengono arrotondati al valore di passo supportato più vicino (ad esempio, da 5 a 6).
Memoria	2G (GB)	128G (GB)	Valori del passo: 2, 3, 4, 5, ..., 128. I valori richiesti vengono arrotondati al valore di passo supportato più vicino (ad esempio, 3,5 G a 4 G).
Archiviazione	-	100G (GB)	Qualsiasi valore. Se vengono richiesti più di 100 GB, ne vengono forniti solo 100 GB.

Per saperne di più sulle unità di risorse in Kubernetes, consulta Unità di risorse in Kubernetes.

L'esempio seguente mostra un'attività che utilizza la specifica del pod worker personalizzata:

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': '0.5',
                            'memory': '2G',
                        })
                    ),
                ],
            ),
        )
    },
)

Visualizza i log delle attività

I log delle attività eseguite da KubernetesExecutor sono disponibili nella scheda Log, insieme ai log delle attività eseguite da CeleryExecutor:

Nella console Google Cloud , vai alla pagina Ambienti.

Vai ad Ambienti
Nell'elenco degli ambienti, fai clic sul nome del tuo ambiente. Viene visualizzata la pagina Dettagli ambiente.
Vai alla scheda Log.
Vai a Tutti i log > Log di Airflow > Worker.
I worker denominati airflow-k8s-worker eseguono le attività di KubernetesExecutor. Per cercare i log di un'attività specifica, puoi utilizzare un ID DAG o un ID attività come parola chiave nella ricerca.