ssiddiqui-apixio edited a comment on issue #12309:
URL: https://github.com/apache/airflow/issues/12309#issuecomment-805276237
sure, @ashb
This is the case which works.
`
from airflow.models import DAG
from airflow.operators.python import task
from airflow.utils.dates import days_ago
from airflow.utils.task_group import TaskGroup
@task
def show():
print("cats are awesome")
with DAG(
"using_task_group",
default_args={'owner': 'airflow'},
start_date=days_ago(2),
schedule_interval=None,
) as dag3:
with TaskGroup(group_id="tasks_1") as tg1:
for _ in range(10):
BashOperator(task_id=f'processing_{_}', bash_command='ls')
`
This case fails. The difference here is that in the taskGroup, I am calling
a task via a method decorator.
`from airflow.models import DAG
from airflow.operators.python import task
from airflow.utils.dates import days_ago
from airflow.utils.task_group import TaskGroup
@task
def show():
print("cats are awesome")
with DAG(
"using_task_group",
default_args={'owner': 'airflow'},
start_date=days_ago(2),
schedule_interval=None,
) as dag3:
with TaskGroup(group_id="tasks_1") as tg1:
for _ in range(10):
show()`
This scenario results in the following error:
>
(airflow) ssiddiqui@siddiquiMac ~ % airflow dags test using_task_group
2021-03-19T19:20:03
[2021-03-23 14:29:32,716] {dagbag.py:448} INFO - Filling up the DagBag from
/Users/ssiddiqui/dev/airflow/spark/data_v2/stg
/Users/ssiddiqui/.pyenv/versions/3.7.3/envs/airflow/lib/python3.7/site-packages/future/standard_library/__init__.py:65
DeprecationWarning: the imp module is deprecated in favour of importlib; see
the module's documentation for alternative uses
[2021-03-23 14:29:33,214] {dagbag.py:305} ERROR - Failed to import:
/Users/ssiddiqui/dev/airflow/spark/data_v2/stg/bug_example.py
Traceback (most recent call last):
File
"/Users/ssiddiqui/.pyenv/versions/3.7.3/envs/airflow/lib/python3.7/site-packages/airflow/models/dagbag.py",
line 302, in _load_modules_from_file
loader.exec_module(new_module)
File "<frozen importlib._bootstrap_external>", line 728, in exec_module
File "<frozen importlib._bootstrap>", line 219, in
_call_with_frames_removed
File "/Users/ssiddiqui/dev/airflow/spark/data_v2/stg/bug_example.py", line
79, in <module>
show()
File
"/Users/ssiddiqui/.pyenv/versions/3.7.3/envs/airflow/lib/python3.7/site-packages/airflow/operators/python.py",
line 294, in factory
**kwargs,
File
"/Users/ssiddiqui/.pyenv/versions/3.7.3/envs/airflow/lib/python3.7/site-packages/airflow/models/baseoperator.py",
line 89, in __call__
obj: BaseOperator = type.__call__(cls, *args, **kwargs)
File
"/Users/ssiddiqui/.pyenv/versions/3.7.3/envs/airflow/lib/python3.7/site-packages/airflow/utils/decorators.py",
line 94, in wrapper
result = func(*args, **kwargs)
File
"/Users/ssiddiqui/.pyenv/versions/3.7.3/envs/airflow/lib/python3.7/site-packages/airflow/operators/python.py",
line 170, in __init__
super().__init__(**kwargs)
File
"/Users/ssiddiqui/.pyenv/versions/3.7.3/envs/airflow/lib/python3.7/site-packages/airflow/utils/decorators.py",
line 94, in wrapper
result = func(*args, **kwargs)
File
"/Users/ssiddiqui/.pyenv/versions/3.7.3/envs/airflow/lib/python3.7/site-packages/airflow/models/baseoperator.py",
line 410, in __init__
task_group.add(self)
File
"/Users/ssiddiqui/.pyenv/versions/3.7.3/envs/airflow/lib/python3.7/site-packages/airflow/utils/task_group.py",
line 140, in add
raise DuplicateTaskIdFound(f"Task id '{key}' has already been added to
the DAG")
airflow.exceptions.DuplicateTaskIdFound: Task id 'tasks_1.show__1' has
already been added to the DAG
>
--
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.
For queries about this service, please contact Infrastructure at:
[email protected]
