GitHub user raphaelauv edited a comment on the discussion: Custom worker images
to lower the number of kubernetespodoperator tasks
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
```python
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)
```
GitHub link:
https://github.com/apache/airflow/discussions/76535#discussioncomment-19417454
----
This is an automatically sent email for [email protected].
To unsubscribe, please send an email to: [email protected]
