You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@airflow.apache.org by GitBox <gi...@apache.org> on 2022/05/19 01:02:43 UTC
[GitHub] [airflow] kevin-woodward opened a new issue, #23792: Dynamic task mapping with KubernetesPodOperator maps too many tasks which fail
kevin-woodward opened a new issue, #23792:
URL: https://github.com/apache/airflow/issues/23792
### Apache Airflow version
2.3.0 (latest released)
### What happened
Excess pods are spawned when using dynamic task mapping, but only in certain cases which I do not understand. I have a simple working example of this where the flow is:
- One pod that returns an XCom of a list (of lists, since I'm partial-ing to `KubernetesPodOperator`'s `arguments`) of length 3. This looks like `[["a"], ["b"], ["c"]]`
- A `partial` from this, which is expanded on the above's result. Each resulting task has an XCom of a single element list that looks like `["d"]`. We expect the `expand` to result in 3 tasks, which it does. So far so good.
- A `partial` from the above. We expect 3 tasks in this final stage, but get 9. 3 succeed and 6 fail consistently. This 3x rule scales to as many tasks as you define in step 2 (e.g. 2 tasks in step 2 -> 6 tasks in step 3, where 4 fail)
If I recreate this using the TaskFlow API with `PythonOperator`s, I get the expected result of 1 task -> 3 tasks -> 3 tasks
Futhermore, if I attempt to look at the `Rendered Template` of the failed tasks in the first image, I consistently get `Error rendering template` and all the fields are `None`.
### What you think should happen instead
I think this has to do with how XComs are specially handled with the `KubernetesPodOperator`. If we look at the XComs tab of the upstream task (`some-task-2` in the above images), we see that the return value specifies `pod_name` and `pod_namespace` along with `return_value`
Whereas in the `t2` task of the TaskFlow version, it only contains `return_value`.
I haven't dug through the code to verify, but I have a strong feeling these extra values `pod_name` and `pod_namespace` are being used to generate the `OperatorPartial`/`MappedOperator` as well when they shouldn't be.
### How to reproduce
Run this DAG in a k8s context:
```
from datetime import datetime
from airflow import XComArg
from airflow.models import DAG
from airflow.providers.cncf.kubernetes.operators.kubernetes_pod import KubernetesPodOperator
def make_operator(
**kwargs
):
return KubernetesPodOperator(
**{
'get_logs': True,
'in_cluster': True,
'is_delete_operator_pod': True,
'namespace': 'default',
'startup_timeout_seconds': 600,
**kwargs,
}
)
def make_partial_operator(
**kwargs
):
return KubernetesPodOperator.partial(
**{
'get_logs': True,
'in_cluster': True,
'is_delete_operator_pod': True,
'namespace': 'default',
'startup_timeout_seconds': 600,
**kwargs,
}
)
with DAG(dag_id='test-pod-xcoms',
schedule_interval=None,
start_date=datetime(2020, 1, 1),
max_active_tasks=20) as dag:
op1 = make_operator(
cmds=['python3', '-c' 'import json;f=open("/airflow/xcom/return.json", "w");f.write(json.dumps([["a"], ["b"], ["c"]]))'],
image='python:3.9-alpine',
name='airflow-private-image-pod-1',
task_id='some-task-1',
do_xcom_push=True
)
op2 = make_partial_operator(
cmds=['python3', '-c' 'import json;f=open("/airflow/xcom/return.json", "w");f.write(json.dumps(["d"]))'],
image='python:3.9-alpine',
name='airflow-private-image-pod-2',
task_id='some-task-2',
do_xcom_push=True
)
op3 = make_partial_operator(
cmds=['echo', 'helloworld'],
image='alpine:latest',
name='airflow-private-image-pod-3',
task_id='some-task-3',
)
op3.expand(arguments=XComArg(op2.expand(arguments=XComArg(op1))))
```
For the TaskFlow version of this that works, run this DAG (doesn't have to be k8s context):
```
from datetime import datetime
from airflow.decorators import task
from airflow.models import DAG, Variable
@task
def t1():
return [[1], [2], [3]]
@task
def t2(val):
return val
@task
def t3(val):
print(val)
with DAG(dag_id='test-mapping',
schedule_interval=None,
start_date=datetime(2020, 1, 1)) as dag:
t3.partial().expand(val=t2.partial().expand(val=t1()))
```
### Operating System
MacOS 11.6.5
### Versions of Apache Airflow Providers
Relevant:
```
apache-airflow-providers-cncf-kubernetes==4.0.1
```
Full:
```
apache-airflow-providers-amazon==3.3.0
apache-airflow-providers-celery==2.1.4
apache-airflow-providers-cncf-kubernetes==4.0.1
apache-airflow-providers-docker==2.6.0
apache-airflow-providers-elasticsearch==3.0.3
apache-airflow-providers-ftp==2.1.2
apache-airflow-providers-google==6.8.0
apache-airflow-providers-grpc==2.0.4
apache-airflow-providers-hashicorp==2.2.0
apache-airflow-providers-http==2.1.2
apache-airflow-providers-imap==2.2.3
apache-airflow-providers-microsoft-azure==3.8.0
apache-airflow-providers-mysql==2.2.3
apache-airflow-providers-odbc==2.0.4
apache-airflow-providers-postgres==4.1.0
apache-airflow-providers-redis==2.0.4
apache-airflow-providers-sendgrid==2.0.4
apache-airflow-providers-sftp==2.6.0
apache-airflow-providers-slack==4.2.3
apache-airflow-providers-sqlite==2.1.3
apache-airflow-providers-ssh==2.4.3
```
### Deployment
Official Apache Airflow Helm Chart
### Deployment details
Docker (Docker Desktop)
- Server Version: 20.10.13
- API Version: 1.41
- Builder: 2
Kubernetes (Docker Desktop)
- Env: docker-desktop
- Context: docker-desktop
- Cluster Name: docker-desktop
- Namespace: default
- Container Runtime: docker
- Version: v1.22.5
Helm:
- version.BuildInfo{Version:"v3.6.3", GitCommit:"d506314abfb5d21419df8c7e7e68012379db2354", GitTreeState:"dirty", GoVersion:"go1.16.5"}
### Anything else
_No response_
### Are you willing to submit PR?
- [ ] Yes I am willing to submit a PR!
### Code of Conduct
- [X] I agree to follow this project's [Code of Conduct](https://github.com/apache/airflow/blob/main/CODE_OF_CONDUCT.md)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] uranusjr closed issue #23792: Dynamic task mapping with KubernetesPodOperator maps too many tasks erroneously
Posted by GitBox <gi...@apache.org>.
uranusjr closed issue #23792: Dynamic task mapping with KubernetesPodOperator maps too many tasks erroneously
URL: https://github.com/apache/airflow/issues/23792
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] Gollum999 commented on issue #23792: Dynamic task mapping with KubernetesPodOperator maps too many tasks erroneously
Posted by GitBox <gi...@apache.org>.
Gollum999 commented on issue #23792:
URL: https://github.com/apache/airflow/issues/23792#issuecomment-1212456285
FWIW this bug can also show up with the TaskFlow API if `multiple_outputs` is explicitly set to `True` in the `@task` decorator, or if the decorated function returns a `dict` (which causes `multiple_outputs` to default to `True`).
A workaround for this bug is to set `multiple_outputs=False` on any mapped tasks that forward the value down to another `.expand`. (e.g. `t2` in OP's example.)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] uranusjr commented on issue #23792: Dynamic task mapping with KubernetesPodOperator maps too many tasks erroneously
Posted by GitBox <gi...@apache.org>.
uranusjr commented on issue #23792:
URL: https://github.com/apache/airflow/issues/23792#issuecomment-1212705438
This happened (already fixed in main) for all tasks that push non-default XComs; both `multiple_outputs` and the KubernetesPodOperator do this.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] uranusjr commented on issue #23792: Dynamic task mapping with KubernetesPodOperator maps too many tasks erroneously
Posted by GitBox <gi...@apache.org>.
uranusjr commented on issue #23792:
URL: https://github.com/apache/airflow/issues/23792#issuecomment-1134719462
I seem to recall there’s a bug on expanding a wrong number of tasks, but can’t find it now. This _may_ have been fixed in main.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org