Airflow: "already registered for DAG" when dynamically generate tasks

user_9824232 0
Viewed 928

Recently I started to use TaskFlow API in some of my dag files where the tasks are being dynamically generated and started to notice (a lot) of warning messages in the logs. Below is a dummy dag file that generates this messages:

from airflow.decorators import dag, task
from airflow.utils.dates import days_ago
from datetime import timedelta
from airflow.operators.dummy import DummyOperator
from airflow.utils.task_group import TaskGroup


NUMS = [1, 2]

default_args = {
    "owner": "henrique",
    "depends_on_past": False,
    "email_on_failure": True,
    "email_on_retry": False,
    "retries": 3,
    "retry_delay": timedelta(minutes=15),
}


def print_id(num: int):
    print(num)
    return num


def run_tests():
    results = []
    for i in NUMS:
        result = task(task_id=f"run_{i}")(print_id)(i)
        results.append(result)

    return results


@task()
def agg(results):
    print(results)


@dag(
    "test_tg",
    default_args=default_args,
    schedule_interval="@once",
    start_date=days_ago(1),
    max_active_runs=1,
)
def test_supervisor():
    task_start = DummyOperator(task_id="task_start")
    task_end = DummyOperator(task_id="task_end")
    groups = []
    for i in NUMS:
        with TaskGroup(group_id=f"{i}_num_group") as tg:
            results = run_tests()
            aggregation = agg(results)

            groups.append(tg)

    task_start >> groups >> task_end


data_dag = test_supervisor()

When running this dag I start to get a lot of the following warning messages:

"[2021-08-24 09:46:46,438] {baseoperator.py:1301} WARNING - Dependency <Task(_PythonDecoratedOperator): 2_num_group.agg>, 2_num_group.run_2 already registered for DAG: test_tg"
"[2021-08-24 09:46:46,438] {baseoperator.py:1301} WARNING - Dependency <Task(_PythonDecoratedOperator): 2_num_group.run_2>, 2_num_group.agg already registered for DAG: test_tg"
"[2021-08-24 09:46:46,437] {baseoperator.py:1301} WARNING - Dependency <Task(_PythonDecoratedOperator): 2_num_group.run_1>, 2_num_group.agg already registered for DAG: test_tg"
"[2021-08-24 09:46:46,413] {baseoperator.py:1301} WARNING - Dependency <Task(_PythonDecoratedOperator): 1_num_group.agg>, 1_num_group.run_1 already registered for DAG: test_tg"
....

And these messages keep appearing even when the DAG is not running and also when it's paused.

Am I doing something wrong in creating the tasks?

Thanks in advance

google-cloud-composer airflow python
1 Answers

I tried your code and works just fine, I don't get any of the mentioned warnings. I'm running Airflow v2.1.2 using the official docker-compose setup.

I found a few issues in Airflow's repo (pr, pr) of older versions related to the messages you are recieving, but those should be solved by now. Try upgrading to latest version of Airflow, that should fix the problem.

Edit:

The following is what I obtained after copy and pasting your code in my running AF:

Graph View:

graph view

Logs:

airflow dags test test_tg 2021-08-24 output:

[2021-08-24 15:59:19,247] {dagbag.py:496} INFO - Filling up the DagBag from /opt/airflow/dags
[2021-08-24 15:59:19,865] {base_executor.py:82} INFO - Adding to queue: ['<TaskInstance: test_tg.task_start 2021-08-24 00:00:00+00:00 [queued]>']
[2021-08-24 15:59:24,991] {taskinstance.py:1302} INFO - Exporting the following env vars:
AIRFLOW_CTX_DAG_OWNER=airflow
AIRFLOW_CTX_DAG_ID=test_tg
AIRFLOW_CTX_TASK_ID=task_start
AIRFLOW_CTX_EXECUTION_DATE=2021-08-24T00:00:00+00:00
AIRFLOW_CTX_DAG_RUN_ID=backfill__2021-08-24T00:00:00+00:00
[2021-08-24 15:59:25,003] {taskinstance.py:1211} INFO - Marking task as SUCCESS. dag_id=test_tg, task_id=task_start, execution_date=20210824T000000, start_date=20210824T120558, end_date=20210824T155925
[2021-08-24 15:59:25,043] {taskinstance.py:1265} INFO - 0 downstream tasks scheduled from follow-on schedule check
[2021-08-24 15:59:25,083] {backfill_job.py:388} INFO - [backfill progress] | finished run 0 of 1 | tasks waiting: 7 | succeeded: 1 | running: 0 | failed: 0 | skipped: 0 | deadlocked: 0 | not ready: 7
[2021-08-24 15:59:25,122] {base_executor.py:82} INFO - Adding to queue: ['<TaskInstance: test_tg.1_num_group.run_1 2021-08-24 00:00:00+00:00 [queued]>']
[2021-08-24 15:59:25,164] {base_executor.py:82} INFO - Adding to queue: ['<TaskInstance: test_tg.1_num_group.run_2 2021-08-24 00:00:00+00:00 [queued]>']
[2021-08-24 15:59:25,241] {base_executor.py:82} INFO - Adding to queue: ['<TaskInstance: test_tg.2_num_group.run_1 2021-08-24 00:00:00+00:00 [queued]>']
[2021-08-24 15:59:25,279] {base_executor.py:82} INFO - Adding to queue: ['<TaskInstance: test_tg.2_num_group.run_2 2021-08-24 00:00:00+00:00 [queued]>']
[2021-08-24 15:59:29,814] {taskinstance.py:1302} INFO - Exporting the following env vars:
AIRFLOW_CTX_DAG_OWNER=airflow
AIRFLOW_CTX_DAG_ID=test_tg
AIRFLOW_CTX_TASK_ID=1_num_group.run_1
AIRFLOW_CTX_EXECUTION_DATE=2021-08-24T00:00:00+00:00
AIRFLOW_CTX_DAG_RUN_ID=backfill__2021-08-24T00:00:00+00:00
1
[2021-08-24 15:59:29,849] {taskinstance.py:1211} INFO - Marking task as SUCCESS. dag_id=test_tg, task_id=1_num_group.run_1, execution_date=20210824T000000, start_date=20210824T120558, end_date=20210824T155929
[2021-08-24 15:59:29,876] {taskinstance.py:1265} INFO - 0 downstream tasks scheduled from follow-on schedule check
[2021-08-24 15:59:29,912] {taskinstance.py:1302} INFO - Exporting the following env vars:
AIRFLOW_CTX_DAG_OWNER=airflow
AIRFLOW_CTX_DAG_ID=test_tg
AIRFLOW_CTX_TASK_ID=1_num_group.run_2
AIRFLOW_CTX_EXECUTION_DATE=2021-08-24T00:00:00+00:00
AIRFLOW_CTX_DAG_RUN_ID=backfill__2021-08-24T00:00:00+00:00
2
[2021-08-24 15:59:29,932] {taskinstance.py:1211} INFO - Marking task as SUCCESS. dag_id=test_tg, task_id=1_num_group.run_2, execution_date=20210824T000000, start_date=20210824T120558, end_date=20210824T155929
[2021-08-24 15:59:29,965] {taskinstance.py:1265} INFO - 0 downstream tasks scheduled from follow-on schedule check
[2021-08-24 15:59:30,027] {taskinstance.py:1302} INFO - Exporting the following env vars:
AIRFLOW_CTX_DAG_OWNER=airflow
AIRFLOW_CTX_DAG_ID=test_tg
AIRFLOW_CTX_TASK_ID=2_num_group.run_1
AIRFLOW_CTX_EXECUTION_DATE=2021-08-24T00:00:00+00:00
AIRFLOW_CTX_DAG_RUN_ID=backfill__2021-08-24T00:00:00+00:00
1
[2021-08-24 15:59:30,060] {taskinstance.py:1211} INFO - Marking task as SUCCESS. dag_id=test_tg, task_id=2_num_group.run_1, execution_date=20210824T000000, start_date=20210824T120558, end_date=20210824T155930
[2021-08-24 15:59:30,087] {taskinstance.py:1265} INFO - 0 downstream tasks scheduled from follow-on schedule check
[2021-08-24 15:59:30,124] {taskinstance.py:1302} INFO - Exporting the following env vars:
AIRFLOW_CTX_DAG_OWNER=airflow
AIRFLOW_CTX_DAG_ID=test_tg
AIRFLOW_CTX_TASK_ID=2_num_group.run_2
AIRFLOW_CTX_EXECUTION_DATE=2021-08-24T00:00:00+00:00
AIRFLOW_CTX_DAG_RUN_ID=backfill__2021-08-24T00:00:00+00:00
2
[2021-08-24 15:59:30,151] {taskinstance.py:1211} INFO - Marking task as SUCCESS. dag_id=test_tg, task_id=2_num_group.run_2, execution_date=20210824T000000, start_date=20210824T120558, end_date=20210824T155930
[2021-08-24 15:59:30,185] {taskinstance.py:1265} INFO - 0 downstream tasks scheduled from follow-on schedule check
[2021-08-24 15:59:30,238] {backfill_job.py:388} INFO - [backfill progress] | finished run 0 of 1 | tasks waiting: 3 | succeeded: 5 | running: 0 | failed: 0 | skipped: 0 | deadlocked: 0 | not ready: 3
[2021-08-24 15:59:30,275] {base_executor.py:82} INFO - Adding to queue: ['<TaskInstance: test_tg.1_num_group.agg 2021-08-24 00:00:00+00:00 [queued]>']
[2021-08-24 15:59:30,310] {base_executor.py:82} INFO - Adding to queue: ['<TaskInstance: test_tg.2_num_group.agg 2021-08-24 00:00:00+00:00 [queued]>']
[2021-08-24 15:59:34,826] {taskinstance.py:1302} INFO - Exporting the following env vars:
AIRFLOW_CTX_DAG_OWNER=airflow
AIRFLOW_CTX_DAG_ID=test_tg
AIRFLOW_CTX_TASK_ID=1_num_group.agg
AIRFLOW_CTX_EXECUTION_DATE=2021-08-24T00:00:00+00:00
AIRFLOW_CTX_DAG_RUN_ID=backfill__2021-08-24T00:00:00+00:00
[1, 2]
[2021-08-24 15:59:34,833] {taskinstance.py:1211} INFO - Marking task as SUCCESS. dag_id=test_tg, task_id=1_num_group.agg, execution_date=20210824T000000, start_date=20210824T120558, end_date=20210824T155934
[2021-08-24 15:59:34,859] {taskinstance.py:1265} INFO - 0 downstream tasks scheduled from follow-on schedule check
[2021-08-24 15:59:34,904] {taskinstance.py:1302} INFO - Exporting the following env vars:
AIRFLOW_CTX_DAG_OWNER=airflow
AIRFLOW_CTX_DAG_ID=test_tg
AIRFLOW_CTX_TASK_ID=2_num_group.agg
AIRFLOW_CTX_EXECUTION_DATE=2021-08-24T00:00:00+00:00
AIRFLOW_CTX_DAG_RUN_ID=backfill__2021-08-24T00:00:00+00:00
[1, 2]
[2021-08-24 15:59:34,915] {taskinstance.py:1211} INFO - Marking task as SUCCESS. dag_id=test_tg, task_id=2_num_group.agg, execution_date=20210824T000000, start_date=20210824T120558, end_date=20210824T155934
[2021-08-24 15:59:34,945] {taskinstance.py:1265} INFO - 0 downstream tasks scheduled from follow-on schedule check
[2021-08-24 15:59:34,982] {backfill_job.py:388} INFO - [backfill progress] | finished run 0 of 1 | tasks waiting: 1 | succeeded: 7 | running: 0 | failed: 0 | skipped: 0 | deadlocked: 0 | not ready: 1
[2021-08-24 15:59:35,014] {base_executor.py:82} INFO - Adding to queue: ['<TaskInstance: test_tg.task_end 2021-08-24 00:00:00+00:00 [queued]>']
[2021-08-24 15:59:39,829] {taskinstance.py:1302} INFO - Exporting the following env vars:
AIRFLOW_CTX_DAG_OWNER=airflow
AIRFLOW_CTX_DAG_ID=test_tg
AIRFLOW_CTX_TASK_ID=task_end
AIRFLOW_CTX_EXECUTION_DATE=2021-08-24T00:00:00+00:00
AIRFLOW_CTX_DAG_RUN_ID=backfill__2021-08-24T00:00:00+00:00
[2021-08-24 15:59:39,841] {taskinstance.py:1211} INFO - Marking task as SUCCESS. dag_id=test_tg, task_id=task_end, execution_date=20210824T000000, start_date=20210824T120558, end_date=20210824T155939
[2021-08-24 15:59:39,867] {taskinstance.py:1265} INFO - 0 downstream tasks scheduled from follow-on schedule check
[2021-08-24 15:59:39,890] {dagrun.py:444} INFO - Marking run <DagRun test_tg @ 2021-08-24 00:00:00+00:00: backfill__2021-08-24T00:00:00+00:00, externally triggered: False> successful
[2021-08-24 15:59:39,898] {backfill_job.py:388} INFO - [backfill progress] | finished run 1 of 1 | tasks waiting: 0 | succeeded: 8 | running: 0 | failed: 0 | skipped: 0 | deadlocked: 0 | not ready: 0
[2021-08-24 15:59:39,905] {backfill_job.py:831} INFO - Backfill done. Exiting.

airflow info output:

default@91172692e679:/opt/airflow$ airflow info

Apache Airflow
version                | 2.1.2                                                 
executor               | CeleryExecutor                                        
task_logging_handler   | airflow.utils.log.file_task_handler.FileTaskHandler   
sql_alchemy_conn       | postgresql+psycopg2://airflow:airflow@postgres/airflow
dags_folder            | /opt/airflow/dags                                     
plugins_folder         | /opt/airflow/plugins                                  
base_log_folder        | /opt/airflow/logs                                     
remote_base_log_folder |                                  
...
...
user_105692200
Related