CODEWITHSUNDEEP
airflowairflow-scheduler
Debodirno Chandra
Debodirno Chandra

Reputation: 183

How to prevent same task name for multiple DAG runs to not collide?

Using Airflow 1.10.3, our DAGs run once everyday. But sometimes, if it spills over, it results in collision of tasks, leading to failure.

Even though the tasks belong to different DAG runs, but since the task name is the same, it results in failure.

[2021-07-27 11:49:09,642] {logging_mixin.py:95} INFO - [<TaskInstance: dag1.part1.task1 2021-07-24 05:30:00+00:00 [running]>, <TaskInstance: dag1.part1.task1 2021-07-26 05:30:00+00:00 [running]>]
[2021-07-27 11:49:09,642] {logging_mixin.py:95} INFO - [2021-07-27 11:49:09,642] {airflow_utils.py:114} INFO - Found existing tasks running with state RUNNING - [<TaskInstance: dag1.part1.task1 2021-07-24 05:30:00+00:00 [running]>, <TaskInstance: dag1.part1.task1 2021-07-26 05:30:00+00:00 [running]>]. Therefore skipping this task instance
[2021-07-27 11:49:09,644] {__init__.py:1580} ERROR - Found existing tasks running with state RUNNING - [<TaskInstance: dag1.part1.task1 2021-07-24 05:30:00+00:00 [running]>, <TaskInstance: dag1.part1.task1 2021-07-26 05:30:00+00:00 [running]>]. Therefore skipping this task instance
Traceback (most recent call last):
  File "/usr/local/lib/python2.7/dist-packages/airflow/models/__init__.py", line 1441, in _run_raw_task
    result = task_copy.execute(context=context)
  File "/usr/local/lib/python2.7/dist-packages/airflow/utils/db.py", line 73, in wrapper
    return func(*args, **kwargs)
  File "/home/ubuntu/airflow/dags/utils/airflow_utils.py", line 115, in execute
    raise RuntimeError(error)
RuntimeError: Found existing tasks running with state RUNNING - [<TaskInstance: dag1.part1.task1 2021-07-24 05:30:00+00:00 [running]>, <TaskInstance: dag1.part1.task1 2021-07-26 05:30:00+00:00 [running]>]. Therefore skipping this task instance

Any idea how to fix this? Thanks in advance.

Upvotes: 0

Views: 940

Answers (1)

Josh Fell
Josh Fell

Reputation: 3589

Based on this entry in the log:

ERROR - Found existing tasks running with state RUNNING ... Therefore skipping this task instance

I suspect you have max_active_runs or max_active_runs_per_dag set to 1. If you need or expect multiple runs of the DAG to execute concurrently you can increase that number accordingly.

Upvotes: 1

Related Questions