Custom worker images to lower the number of kubernetespodoperator tasks

[karjaljo]

I've been running Airflow on distributed virtual machine environment and now I'm moving to OpenShift. Big chunk of my tasks are run using DockerOperator in the VM environment and I have switched those to be KubernetesPodOperator tasks. My processing is time sensitive so I can't use too strict pool sizes or other limiting conditions. Also, many of the pod runs are short lived, only few seconds. This leads to problems within the OpenShift cluster, since it can't handle too many pods too fast.

One solution I was thinking, would be to build custom worker images that contain one or more applications. Since the workers would be always running and I could convert the KubernetesPodOperator tasks to regular tasks, the pod count would stay low. Is this possible and is there a better solution to this problem? So far it has been quite nice to have separate images for each application and building workers with multiple installations would also feel like a messy way to tackle this issue.

[itzmitto]

Hallo @karjaljo,

Yes, this is possible with CeleryExecutor. You can keep separate worker images for different applications rather than install everything into one image.

For example:

• Build an Airflow worker image containing application A and its dependencies.
• Run a persistent worker group listening to an app_a queue with airflow celery worker -q app_a.
• Run that application's commands through ordinary BashOperator tasks assigned to queue="app_a".

You can repeat that setup for another application and queue 😀 Those tasks execute on the existing Celery workers, avoiding a separate KubernetesPodOperator pod for each application invocation.

Each image still needs Airflow and the appropriate configuration and DAG access; an application-only image isn't enough.

The documented approach is here:
https://airflow.apache.org/docs/apache-airflow/stable/best-practices.html#using-multiple-docker-images-and-celery-queues

Queue routing:
https://airflow.apache.org/docs/apache-airflow-providers-celery/stable/celery_executor.html#queues

The trade-off is maintaining multiple worker deployments. The docs warn about the extra configuration and resource-management complexity.

Also, KubernetesExecutor still creates a pod per task, so changing operators alone wouldn't remove that overhead if you're using it.

Which Airflow version and executor are you deploying?

Hope this helps!

[karjaljo]

Thanks a lot for the fast answer! In the virtual machine setup I use airflow 2.11.2 with CeleryExecutor and in OpenShift I use 3.3.1 also with CeleryExecutor.

[raphaelauv]

using airflow celery workers with dockeroperator or ExternalPythonOperator will make it works BUT you will face the problem of managing concurrent task on a airflow_celery_workers where kubernetes is way "smarter"

so you could go for a new operator , ask an LLM this :

In Airflow 3.3, I need to run code that is incompatible with Airflow, in its own image, without Kubernetes creating a new pod for every task.

Set up Celery workers autoscaled by KEDA (scale to zero when idle), with my app image as a sidecar container in each worker pod and worker_concurrency set to 1 (one task per pod). Write a SidecarExecOperator that runs the task's command in that sidecar container with live logs and fails on a non-zero exit code.

Give me the Helm values, the operator, an example DAG, the RBAC, and the limitations.

something lke

import socket
from pathlib import Path
from airflow.providers.cncf.kubernetes.operators.pod_exec import KubernetesPodExecOperator


class SelfPodExecOperator(KubernetesPodExecOperator):
    """Exec into a sibling container of the worker pod running this task."""

    def __init__(self, **kwargs):
        kwargs.setdefault("pod_name", "self")          # placeholder, resolved at runtime
        kwargs.setdefault("namespace", "default")
        super().__init__(**kwargs)

    def execute(self, context):
        self.pod_name = socket.gethostname()           # pod hostname == pod name
        self.namespace = Path(
            "/var/run/secrets/kubernetes.io/serviceaccount/namespace"
        ).read_text().strip()
        return super().execute(context)