Proper way to create dynamic workflows in Airflow

by

Here is how I did it with a similar request without any subdags:

First create a method that returns whatever values you want

def values_function():
     return values

Next create method that will generate the jobs dynamically:

def group(number, **kwargs):
        #load the values if needed in the command you plan to execute
        dyn_value = "{{ task_instance.xcom_pull(task_ids="push_func") }}"
        return BashOperator(
                task_id='JOB_NAME_{}'.format(number),
                bash_command='script.sh {} {}'.format(dyn_value, number),
                dag=dag)

And then combine them:

push_func = PythonOperator(
        task_id='push_func',
        provide_context=True,
        python_callable=values_function,
        dag=dag)

complete = DummyOperator(
        task_id='All_jobs_completed',
        dag=dag)

for i in values_function():
        push_func >> group(i) >> complete

More Related Contents:

  1. Airflow “This DAG isnt available in the webserver DagBag object “
  2. Importing an ipynb file from another ipynb file?
  3. setting up s3 for logs in airflow
  4. How to create a conditional task in Airflow
  5. Importing modules in Python – best practice
  6. How to run Airflow on Windows
  7. pass task instance variables as op_args
  8. How to prevent airflow from backfilling dag runs?
  9. How to control the parallelism or concurrency of an Airflow installation?
  10. Apache Airflow scheduler does not trigger DAG at schedule time
  11. How to create a django ViewFlow process programmatically
  12. Print elements of list horizontally [closed]
  13. How to import a module given its name as string?
  14. “Ask forgiveness not permission” – explain
  15. Difference between __getattr__ and __getattribute__
  16. Reloading submodules in IPython
  17. Error while installing lxml through pip: Microsoft Visual C++ 14.0 is required
  18. Making the background move sideways in pygame
  19. Explicitly select items from a list or tuple
  20. Printing without newline (print ‘a’,) prints a space, how to remove?
  21. ERROR:gpu_process_transport_factory.cc(1007)-Lost UI shared context : while initializing Chrome browser through ChromeDriver in Headless mode
  22. Cross-validation metrics in scikit-learn for each data split
  23. Why can’t I use a list as a dict key in python? Exactly what can and cannot be used, and why?
  24. How can I collect the results of a repeated calculation in a list, dictionary etc. (make a copy of a list with each element modified)?
  25. Pandas: drop columns with all NaN’s
  26. What is the difference between setUp() and setUpClass() in Python unittest?
  27. How can I convert string values from a dictionary, into int/float datatypes?
  28. Embedding Python in C, linking fails with undefined reference to `Py_Initialize’
  29. Is possible to mapping view with class using mapper in SqlAlchemy?
  30. How can I drag more than 2 images in PyGame?

Leave a Comment