You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@airflow.apache.org by "phi-friday (via GitHub)" <gi...@apache.org> on 2024/08/19 08:26:30 UTC
[I] Cannot execute Dynamic Task Mapping when `default_arg` is present. [airflow]
phi-friday opened a new issue, #41578:
URL: https://github.com/apache/airflow/issues/41578
### Apache Airflow version
2.10.0rc1
### If "Other Airflow 2 version" selected, which one?
2.10.0(not rc)
### What happened?
Cannot execute Dynamic Task Mapping when `default_arg` is present.
```shell
dddbfe3b933a
*** Found local files:
*** * /opt/airflow/logs/dag_id=dynamic_task_error/run_id=manual__2024-08-19T08:19:45.119366+00:00/task_id=add/map_index=0/attempt=1.log
[2024-08-19, 17:19:47 KST] {local_task_job_runner.py:123} ▼ Pre task execution logs
[2024-08-19, 17:19:47 KST] {taskinstance.py:2603} INFO - Dependencies all met for dep_context=non-requeueable deps ti=<TaskInstance: dynamic_task_error.add manual__2024-08-19T08:19:45.119366+00:00 map_index=0 [queued]>
[2024-08-19, 17:19:47 KST] {taskinstance.py:2603} INFO - Dependencies all met for dep_context=requeueable deps ti=<TaskInstance: dynamic_task_error.add manual__2024-08-19T08:19:45.119366+00:00 map_index=0 [queued]>
[2024-08-19, 17:19:47 KST] {taskinstance.py:2856} INFO - Starting attempt 1 of 1
[2024-08-19, 17:19:47 KST] {taskinstance.py:2879} INFO - Executing <Mapped(_PythonDecoratedOperator): add> on 2024-08-19 08:19:45.119366+00:00
[2024-08-19, 17:19:47 KST] {warnings.py:112} WARNING - /home/***/.local/lib/python3.12/site-packages/***/task/task_runner/standard_task_runner.py:70: DeprecationWarning: This process (pid=18581) is multi-threaded, use of fork() may lead to deadlocks in the child.
pid = os.fork()
[2024-08-19, 17:19:47 KST] {standard_task_runner.py:72} INFO - Started process 18604 to run task
[2024-08-19, 17:19:47 KST] {standard_task_runner.py:104} INFO - Running: ['***', 'tasks', 'run', 'dynamic_task_error', 'add', 'manual__2024-08-19T08:19:45.119366+00:00', '--job-id', '4088', '--raw', '--subdir', 'DAGS_FOLDER/dynamic_test.py', '--cfg-path', '/tmp/tmpz8eds5gr', '--map-index', '0']
[2024-08-19, 17:19:47 KST] {standard_task_runner.py:105} INFO - Job 4088: Subtask add
[2024-08-19, 17:19:47 KST] {task_command.py:467} INFO - Running <TaskInstance: dynamic_task_error.add manual__2024-08-19T08:19:45.119366+00:00 map_index=0 [running]> on host dddbfe3b933a
[2024-08-19, 17:19:47 KST] {taskinstance.py:3301} ERROR - Task failed with exception
Traceback (most recent call last):
File "/home/airflow/.local/lib/python3.12/site-packages/airflow/models/taskinstance.py", line 273, in _run_raw_task
TaskInstance._execute_task_with_callbacks(
File "/home/airflow/.local/lib/python3.12/site-packages/airflow/models/taskinstance.py", line 3105, in _execute_task_with_callbacks
task_orig = self.render_templates(context=context, jinja_env=jinja_env)
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
File "/home/airflow/.local/lib/python3.12/site-packages/airflow/models/taskinstance.py", line 3524, in render_templates
original_task.render_template_fields(context, jinja_env)
File "/home/airflow/.local/lib/python3.12/site-packages/airflow/models/mappedoperator.py", line 913, in render_template_fields
unmapped_task = self.unmap(mapped_kwargs)
^^^^^^^^^^^^^^^^^^^^^^^^^
File "/home/airflow/.local/lib/python3.12/site-packages/airflow/models/mappedoperator.py", line 824, in unmap
op = self.operator_class(**kwargs, _airflow_from_mapped=True)
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
File "/home/airflow/.local/lib/python3.12/site-packages/airflow/models/baseoperator.py", line 490, in apply_defaults
result = func(self, **kwargs, default_args=default_args)
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
File "/home/airflow/.local/lib/python3.12/site-packages/airflow/decorators/python.py", line 52, in __init__
super().__init__(
File "/home/airflow/.local/lib/python3.12/site-packages/airflow/models/baseoperator.py", line 490, in apply_defaults
result = func(self, **kwargs, default_args=default_args)
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
File "/home/airflow/.local/lib/python3.12/site-packages/airflow/decorators/base.py", line 258, in __init__
super().__init__(task_id=task_id, **kwargs_to_upstream, **kwargs)
File "/home/airflow/.local/lib/python3.12/site-packages/airflow/models/baseoperator.py", line 490, in apply_defaults
result = func(self, **kwargs, default_args=default_args)
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
File "/home/airflow/.local/lib/python3.12/site-packages/airflow/operators/python.py", line 222, in __init__
super().__init__(**kwargs)
File "/home/airflow/.local/lib/python3.12/site-packages/airflow/models/baseoperator.py", line 490, in apply_defaults
result = func(self, **kwargs, default_args=default_args)
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
File "/home/airflow/.local/lib/python3.12/site-packages/airflow/models/baseoperator.py", line 938, in __init__
raise AirflowException(
airflow.exceptions.AirflowException: Invalid arguments were passed to _PythonDecoratedOperator (task_id: add__1). Invalid arguments were:
**kwargs: {'key': 'value', 'arbitrary': 'value'}
[2024-08-19, 17:19:47 KST] {taskinstance.py:3349} ERROR - Unable to unmap task to determine if we need to send an alert email
[2024-08-19, 17:19:47 KST] {taskinstance.py:1225} INFO - Marking task as FAILED. dag_id=dynamic_task_error, task_id=add, run_id=manual__2024-08-19T08:19:45.119366+00:00, map_index=0, execution_date=20240819T081945, start_date=20240819T081947, end_date=20240819T081947
[2024-08-19, 17:19:47 KST] {taskinstance.py:340} ▶ Post task execution logs
```
### What you think should happen instead?
Unused arguments should be ignored (i.e., they should behave the same as any other task).
As you can see in the example below, `Task <success>` is executed, but `Task <added_values>` is not.
### How to reproduce
```python
from __future__ import annotations
from airflow.decorators import dag, task
from pendulum import datetime
@dag(
start_date=datetime(2024, 1, 1),
schedule=None,
catchup=False,
default_args={"key": "value", "arbitrary": "value", "do_xcom_push": False},
)
def dynamic_task_error() -> None:
@task.python()
def success() -> str:
return "success"
@task.python(do_xcom_push=True)
def add(x: int, y: int) -> int:
return x + y
do_success = success()
added_values = add.partial(y=10).expand(x=[1, 2, 3])
_ = do_success >> added_values
dynamic_task_error()
```
### Operating System
PRETTY_NAME="Debian GNU/Linux 12 (bookworm)" NAME="Debian GNU/Linux" VERSION_ID="12" VERSION="12 (bookworm)" VERSION_CODENAME=bookworm ID=debian HOME_URL="https://www.debian.org/" SUPPORT_URL="https://www.debian.org/support" BUG_REPORT_URL="https://bugs.debian.org/"
### Versions of Apache Airflow Providers
apache-airflow-providers-celery==3.7.3
apache-airflow-providers-common-compat==1.1.0
apache-airflow-providers-common-io==1.4.0
apache-airflow-providers-common-sql==1.15.0
apache-airflow-providers-docker==3.12.3
apache-airflow-providers-fab==1.2.2
apache-airflow-providers-ftp==3.10.1
apache-airflow-providers-http==4.12.0
apache-airflow-providers-imap==3.6.1
apache-airflow-providers-jdbc==4.4.0
apache-airflow-providers-odbc==4.6.3
apache-airflow-providers-postgres==5.11.3
apache-airflow-providers-redis==3.7.1
apache-airflow-providers-smtp==1.7.1
apache-airflow-providers-sqlite==3.8.2
### Deployment
Docker-Compose
### Deployment details
_No response_
### 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
Re: [I] Cannot execute Dynamic Task Mapping when `default_arg` is present. [airflow]
Posted by "phi-friday (via GitHub)" <gi...@apache.org>.
phi-friday closed issue #41578: Cannot execute Dynamic Task Mapping when `default_arg` is present.
URL: https://github.com/apache/airflow/issues/41578
--
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
Re: [I] Cannot execute Dynamic Task Mapping when `default_arg` is present. [airflow]
Posted by "eladkal (via GitHub)" <gi...@apache.org>.
eladkal commented on issue #41578:
URL: https://github.com/apache/airflow/issues/41578#issuecomment-2296009190
Isn't that duplicate of https://github.com/apache/airflow/issues/33600 ?
--
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
Re: [I] Cannot execute Dynamic Task Mapping when `default_arg` is present. [airflow]
Posted by "phi-friday (via GitHub)" <gi...@apache.org>.
phi-friday commented on issue #41578:
URL: https://github.com/apache/airflow/issues/41578#issuecomment-2296022674
> Isn't that duplicate of #33600 ?
It's the exact same problem. Close the issue due to duplication.
--
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