4
\$\begingroup\$

I would like to know if what I did to achieve to goal of dynamic operators within an Airflow DAG (Directed Acyclic Graph) is a good or a bad practice.

The goal I had to achieve was:

Create a 'x' amount of operators within a DAG based on the result of an API call. This DAG will run for example every week.

The program that I made works, but I don't know if it is a good practice for developing DAGs for Airflow. So my question is: is it a good practice or if not what would be a better solution to this problem?

The code I used to achieve this goal:

with DAG('my_dag', default_args=default_args, schedule_interval='@weekly') as my_dag:

    start = DummyOperator(
        task_id='start'
    )

    endpoint = 'www.example.com/api/dummies'
    r = requests.get(endpoint)
    dummies = r.json()

   for _, dummy in enumerate(dummies):
        dummy_operator = DummyOperator(
                    task_id='dummy_opr_{}'.format(dummy['id']
                    )
        start >> dummy_operator
\$\endgroup\$
1
\$\begingroup\$

It is a good practice for Airflow. Moreover, Airflow positioning it as one of their "killer-features". It is good to create dynamic tasks in DAG, it is pretty OK to create dynamically created DAGs. One thing I recommend you to think about is that it is not good to create hundreds and thousands of one-time-running tasks because it will lead to:

  1. Filling your Airflow database with huge amount of garbage data
  2. Inability to analyze DAGs visually. Here is the example of Airflow tree tab of tasks like these:

enter image description here

Here is the sample code for dynamic DAG creation:

from datetime import datetime
from airflow import DAG
from airflow.operators.python_operator import PythonOperator


def create_dag(dag_id, schedule, dag_number, default_args):

    def hello_world_py(*args):
        print('Hello World')
        print('This is DAG: {}'.format(str(dag_number)))

    dag = DAG(dag_id,
              schedule_interval=schedule,
              default_args=default_args)

    with dag:
        t1 = PythonOperator(
            task_id='hello_world',
            python_callable=hello_world_py,
            dag_number=dag_number)

    return dag


# build a dag for each number in range(10)
for n in range(1, 10):
    dag_id = 'hello_world_{}'.format(str(n))

    default_args = {'owner': 'airflow',
                    'start_date': datetime(2018, 1, 1)
                    }

    schedule = '@daily'

    dag_number = n

    globals()[dag_id] = create_dag(dag_id,
                                  schedule,
                                  dag_number,
                                  default_args)

You can read more about it here.

\$\endgroup\$
  • \$\begingroup\$ Thanks for this answer. I would like to know if this is still a good practice if a replace your for n in range(1,10) with the data of an API call as I do in my solution. But than I wouldn't create a lot of tasks within a dag. I would create a new DAG for every item returned by the API? \$\endgroup\$ – Mar-k May 24 at 12:35
  • \$\begingroup\$ I think create a DAG for each API call is a bad idea (generally, I don't know what exactly is in your project). Instead you can create a task in DAG for each API call. \$\endgroup\$ – vurmux May 24 at 13:39
  • \$\begingroup\$ What I do in my example above is create a task in a dag for each item returned by an API call. In my case we have items returned by an API call which have a start date and a schedule_interval so I think I am best of using your solution :) \$\endgroup\$ – Mar-k May 27 at 7:34

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service, privacy policy and cookie policy

Not the answer you're looking for? Browse other questions tagged or ask your own question.