Menggunakan CeleryKubernetesExecutor

Managed Airflow (Gen 3) | Managed Airflow (Gen 2) | Managed Airflow (Legacy Gen 1)

Halaman ini menjelaskan cara mengaktifkan CeleryKubernetesExecutor di Managed Airflow dan cara menggunakan KubernetesExecutor di DAG Anda.

Tentang CeleryKubernetesExecutor

CeleryKubernetesExecutor adalah jenis executor yang dapat menggunakan CeleryExecutor dan KubernetesExecutor secara bersamaan. Airflow memilih executor berdasarkan antrean yang Anda tentukan untuk tugas. Dalam satu DAG, Anda dapat menjalankan beberapa tugas dengan CeleryExecutor, dan tugas lainnya dengan KubernetesExecutor:

  • CeleryExecutor dioptimalkan untuk eksekusi tugas yang cepat dan skalabel.
  • KubernetesExecutor dirancang untuk eksekusi tugas yang menggunakan banyak resource dan menjalankan tugas secara terpisah.

CeleryKubernetesExecutor di Managed Airflow

CeleryKubernetesExecutor di Managed Airflow memberikan kemampuan untuk menggunakan KubernetesExecutor untuk tugas Anda. Anda tidak dapat menggunakan KubernetesExecutor di Managed Airflow secara terpisah dari CeleryKubernetesExecutor.

Managed Airflow menjalankan tugas yang Anda jalankan dengan KubernetesExecutor di cluster lingkungan Anda, di namespace yang sama dengan pekerja Airflow. Tugas tersebut memiliki binding yang sama dengan pekerja Airflow dan dapat mengakses resource di project Anda.

Tugas yang Anda jalankan dengan KubernetesExecutor menggunakan model harga Managed Airflow, karena pod dengan tugas ini berjalan di cluster lingkungan Anda. SKU Compute Managed Airflow (untuk CPU, Memori, dan Penyimpanan) berlaku untuk pod ini.

Sebaiknya jalankan tugas dengan CeleryExecutor jika:

  • Waktu startup tugas penting.
  • Tugas tidak memerlukan isolasi runtime dan tidak menggunakan banyak resource.

Sebaiknya jalankan tugas dengan KubernetesExecutor jika:

  • Tugas memerlukan isolasi runtime. Misalnya, agar tugas tidak bersaing untuk memperebutkan memori dan CPU, karena tugas berjalan di podnya sendiri.
  • Tugas menggunakan banyak resource dan Anda ingin mengontrol resource CPU dan memori yang tersedia.

KubernetesExecutor dibandingkan dengan KubernetesPodOperator

Menjalankan tugas dengan KubernetesExecutor mirip dengan menjalankan tugas menggunakan KubernetesPodOperator. Tugas dijalankan di pod, sehingga memberikan isolasi tugas tingkat pod dan pengelolaan resource yang lebih baik.

Namun, ada beberapa perbedaan utama:

  • KubernetesExecutor hanya menjalankan tugas di namespace Managed Airflow versi lingkungan Anda. Anda tidak dapat mengubah namespace ini di Managed Airflow. Anda dapat menentukan namespace tempat KubernetesPodOperator menjalankan tugas pod.
  • KubernetesExecutor dapat menggunakan operator Airflow bawaan apa pun. KubernetesPodOperator hanya menjalankan skrip yang disediakan yang ditentukan oleh titik entri container.
  • KubernetesExecutor menggunakan image Docker Managed Airflow default dengan Python, penggantian opsi konfigurasi Airflow, variabel lingkungan, dan paket PyPI yang sama yang ditentukan di lingkungan Managed Airflow Anda.

Tentang image Docker

Secara default, KubernetesExecutor meluncurkan tugas menggunakan image Docker yang sama yang digunakan Managed Airflow untuk pekerja Celery. Ini adalah image Managed Airflow untuk lingkungan Anda, dengan semua perubahan yang Anda tentukan untuk lingkungan Anda, seperti paket PyPI kustom atau variabel lingkungan.

Sebelum memulai

  • Anda dapat menggunakan CeleryKubernetesExecutor di Managed Airflow (Gen 3).

  • Anda tidak dapat menggunakan executor selain CeleryKubernetesExecutor di Managed Airflow (Gen 3). Artinya, Anda dapat menjalankan tugas menggunakan CeleryExecutor, KubernetesExecutor, atau keduanya dalam satu DAG, tetapi Anda tidak dapat mengonfigurasi lingkungan untuk hanya menggunakan KubernetesExecutor atau CeleryExecutor.

Mengonfigurasi CeleryKubernetesExecutor

Anda mungkin ingin mengganti opsi konfigurasi Airflow yang ada yang terkait dengan KubernetesExecutor:

  • [kubernetes]worker_pods_creation_batch_size

    Opsi ini menentukan jumlah panggilan pembuatan Pod Pekerja Kubernetes per loop penjadwal. Nilai defaultnya adalah 1, sehingga hanya satu pod yang diluncurkan per detak jantung penjadwal. Jika Anda menggunakan KubernetesExecutor secara intensif, sebaiknya tingkatkan nilai ini.

  • [kubernetes]worker_pods_pending_timeout

    Opsi ini menentukan, dalam detik, berapa lama pekerja dapat tetap dalam status Pending (Pod sedang dibuat) sebelum dianggap gagal. Nilai defaultnya adalah 5 menit.

Menjalankan tugas dengan KubernetesExecutor atau CeleryExecutor

Anda dapat menjalankan tugas menggunakan CeleryExecutor, KubernetesExecutor, atau keduanya dalam satu DAG:

Airflow 3

  • Untuk menjalankan tugas dengan KubernetesExecutor, tentukan: nilai KubernetesExecutor dalam parameter executor tugas.
  • Untuk menjalankan tugas dengan CeleryExecutor, hapus parameter executor.

Airflow 2

  • Untuk menjalankan tugas dengan KubernetesExecutor, tentukan: nilai kubernetes dalam parameter queue tugas.
  • Untuk menjalankan tugas dengan CeleryExecutor, hapus parameter queue.

Contoh berikut menjalankan tugas task-kubernetes menggunakan KubernetesExecutor dan tugas task-celery menggunakan CeleryExecutor:

Airflow 3 

import datetime
import airflow
from airflow.providers.standard.operators.python import PythonOperator

with airflow.DAG(
  "composer_sample_celery_kubernetes",
  start_date=datetime.datetime(2026, 1, 1),
  schedule=None) 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,
    executor='KubernetesExecutor')

  # 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

Airflow 2 

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

with airflow.DAG(
  "composer_sample_celery_kubernetes",
  start_date=datetime.datetime(2026, 1, 1),
  schedule=None) 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

Menjalankan perintah Airflow CLI yang terkait dengan KubernetesExecutor

Anda dapat menjalankan beberapa perintah Airflow CLI yang terkait dengan KubernetesExecutor menggunakan gcloud.

Menyesuaikan spesifikasi pod pekerja

Anda dapat menyesuaikan spesifikasi pod pekerja dengan meneruskannya dalam parameter executor_config tugas. Anda dapat menggunakan parameter ini untuk menentukan persyaratan CPU dan memori kustom.

Anda dapat mengganti seluruh spesifikasi pod pekerja yang digunakan untuk menjalankan tugas. Untuk mengambil spesifikasi pod tugas yang digunakan oleh KubernetesExecutor, Anda dapat menjalankan kubernetes generate-dag-yaml Airflow CLI perintah.

Untuk mengetahui informasi selengkapnya tentang cara menyesuaikan spesifikasi pod pekerja, lihat dokumentasi Airflow.

Managed Airflow (Gen 3) mendukung nilai berikut untuk persyaratan resource:

Resource Minimum Maksimum Langkah
CPU 0,25 32 Nilai langkah: 0,25, 0,5, 1, 2, 4, 6, 8, 10, ..., 32. Nilai yang diminta dibulatkan ke atas ke nilai langkah yang didukung terdekat (misalnya, 5 menjadi 6).
Memori 2G (GB) 128G (GB) Nilai langkah: 2, 3, 4, 5, ..., 128. Nilai yang diminta dibulatkan ke atas ke nilai langkah yang didukung terdekat (misalnya, 3,5G menjadi 4G).
Penyimpanan - 100G (GB) Semua nilai. Jika lebih dari 100 GB diminta, hanya 100 GB yang disediakan.

Untuk mengetahui informasi selengkapnya tentang unit resource di Kubernetes, lihat Unit resource di Kubernetes.

Contoh berikut menunjukkan tugas yang menggunakan spesifikasi pod pekerja kustom:

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

Melihat log tugas

Log tugas yang dijalankan oleh KubernetesExecutor tersedia di tab Logs , bersama dengan log tugas yang dijalankan oleh CeleryExecutor:

  1. Di Google Cloud konsol, buka halaman Environments.

    Buka Environments

  2. Dalam daftar lingkungan, klik nama lingkungan Anda. Halaman Environment details akan terbuka.

  3. Buka tab Logs.

  4. Buka All logs > Airflow logs > Workers.

  5. Pekerja bernama airflow-k8s-worker menjalankan tugas KubernetesExecutor. Untuk mencari log tugas tertentu, Anda dapat menggunakan ID DAG atau ID tugas sebagai kata kunci dalam penelusuran.

Langkah berikutnya