Use DAG Factory to create DAGs

The DAG Factory is an open source tool managed by Astronomer that allows you to dynamically generate Apache Airflow® DAGs from YAML. While Airflow DAGs are traditionally written exclusively in Python, the DAG Factory makes it easy for people who don’t know Python to use Airflow.

This guide includes instructions for installing the DAG Factory package into your Astro project and a sample YAML configuration file that you can use to easily specify the details of your DAG, including its schedule, callbacks, and task names.

The DAG Factory can be used with all Astronomer products and any Apache Airflow installation. To view the source code of the project, see DAG Factory.

Prerequisites

• Python version 3.8.0 or greater
• Apache Airflow version 2.0 or greater

If you’re an Astronomer customer, you must have an Astro project with a supported version of Astro Runtime. See Astro Runtime lifecycle schedule for a list of currently supported Runtime versions.

Step 1: Install DAG Factory

To use the DAG Factory, install it as a Python package into your Apache Airflow environment.

If you’re an Astronomer customer:

1. In your Astro project, open your requirements.txt file.
2. Add dag-factory<=1.0.0 to the file.
3. Save the changes to your requirements.txt file.

If you’re not an Astronomer customer, install the Python package according to the standards at your organization:

1
pip install dag-factory<=1.0.0

Step 2: Create a sub-directory for your DAG configuration files and create a YAML file for your DAG

1. In the dags directory of your Astro project, create a new sub-directory called configs to store your DAG configuration files defined in YAML. Astronomer recommends keeping these separate from standard DAG files written in Python.
2. Within the new sub-directory, create a new YAML file for your DAG called for example my_dag.yaml. Copy the contents of the following example configuration file into the YAML file.

1
<your-DAG-id>:
2
  default_args:
3
    owner: 'example_owner'
4
    start_date: 2024-07-01  # or '2 days'
5
    end_date: 2030-01-01
6
    retries: 1
7
    retry_delay_sec: 300
8
  schedule_interval: '0 3 * * *'
9
  catchup: False
10
  concurrency: 1
11
  max_active_runs: 1
12
  dagrun_timeout_sec: 60
13
  default_view: 'grid'  # or 'graph', 'duration', 'gantt', 'landing_times' (for Run Duration in newer versions)
14
  orientation: 'LR'  # or 'TB', 'RL', 'BT'
15
  description: 'This is an example dag!'
16
  tasks:
17
    task_1:
18
      operator: airflow.operators.bash_operator.BashOperator
19
      bash_command: 'echo 1'
20
    task_2:
21
      operator: airflow.operators.bash_operator.BashOperator
22
      bash_command: 'echo 2'
23
      dependencies: [task_1]
24
    task_3:
25
      operator: airflow.operators.bash_operator.BashOperator
26
      bash_command: 'echo 3'
27
      dependencies: [task_1]

3. Modify the example configuration file with parameters for the DAG you want to create, including replacing <your-DAG-id> with a valid dag_id. See DAG-level parameters in Airflow to learn more about each parameter.

4. (Optional) Delete any configurations that you don’t want to specify. For parameters that aren’t specified, your DAG will assume the default values that correspond with your current Apache Airflow or Astro Runtime version.

Step 3: Create a DAG Factory file

All YAML files in your dags directory must be parsed and converted into Python in order to run on Apache Airflow. In this step, you will create a new DAG Factory file in your Astro project that includes the conversion logic. You only need to do this once and do not need a separate DAG Factory file for each of your DAGs or YAML files.

1. In the dags directory of your Astro project, create a new Python file called dag_factory.py.
2. Copy the following contents into your empty Python file. This file represents an Apache Airflow DAG and includes two commands that convert each of your YAML file(s) into DAGs.

1
from pathlib import Path
2
3
from airflow import DAG
4
from airflow.configuration import conf as airflow_conf
5
from dagfactory import load_yaml_dags
6
7
config_dir = Path(airflow_conf.get("core", "dags_folder")) / "configs"
8
load_yaml_dags(globals_dict=globals(), dags_folder=config_dir)

Step 4: (Optional) Add a DAG-level callback

In order to use DAG-level callbacks you will need to add callback parameters to your config file. The values will be the paths to the files that contain your callback functions, as well as the callback function names.

In my_dag.yml, add the following parameters:

1
 on_success_callback_name: placeholder_callback
2
 on_success_callback_file: /usr/local/airflow/dags/callback_func.py
3
 on_failure_callback_name: placeholder_callback
4
 on_failure_callback_file: /usr/local/airflow/dags/callback_func.py

1. Create a new file callback_func.py in your dags directory.
2. Copy the contents of the following placeholder callback into the file:

1
def placeholder_callback():
2
    pass

3. Save the file.

Generating YAML files dynamically

The example above shows how to use the DAG factory to create DAGs based on static YAML files. For use cases where you’d like to create several DAGs with a similar structure it is possible to create them dynamically based on a template YAML file to avoid code duplication. Creating a DAG dynamically with the DAG factory simply means that you use Python code to create the YAML configurations instead of writing them manually.

There are two files that you need:

• A template YAML file that contains the structure of the DAGs you want to create with placeholders for the values that will change.
• A Python script that creates the DAG Factory YAML file by replacing the placeholders in the template YAML file with the actual values.

The template YAML file provides the structure for all the DAGs you will generate dynamically with placeholders for values that vary in between the DAGs:

The Python script reads the template YAML file, replaces the placeholders with the actual values, and writes the resulting YAML files to the dags directory. You can run this script manually to generate your DAGs for local development or automatically as part of your CI/CD pipeline.

1
"""
2
generate_dynamic_dag.py
3
4
Script that is run to generate a .yml file to be used to source a dag-factory DAG.
5
"""
6
7
import json
8
9
import yaml
10
11
# provide the path to your template YAML file here
12
PATH_TO_TEMPLATE_YAML: str = "include/template.yml"
13
14
# read the template dag-factory YAML file
15
with open(PATH_TO_TEMPLATE_YAML, "r") as yaml_file:
16
    template: dict = yaml.load(yaml_file, yaml.SafeLoader)
17
18
# replace the placeholders with the actual values,
19
# note that << >> is used to identify placeholders here, this is arbitrary, you
20
# can use replace any other character or sequence of characters
21
TEMPLATE_VARIABLES: list = [
22
    {
23
        "<< dag_id >>": "business_analytics",
24
        "<< default_args_retries >>": 1,
25
        "<< schedule >>": "@daily",
26
        "<< bash_command_task_1 >>": "echo 'Task 1: Business Analytics'",
27
        "<< bash_command_task_2 >>": "echo 'Task 2: Business Analytics'",
28
        "<< bash_command_task_3 >>": "echo 'Task 3: Business Analytics'",
29
    },
30
    {
31
        "<< dag_id >>": "data_science",
32
        "<< default_args_retries >>": 2,
33
        "<< schedule >>": "@weekly",
34
        "<< bash_command_task_1 >>": "echo 'Task 1: Data Science'",
35
        "<< bash_command_task_2 >>": "echo 'Task 2: Data Science'",
36
        "<< bash_command_task_3 >>": "echo 'Task 3: Data Science'",
37
    },
38
    {
39
        "<< dag_id >>": "machine_learning",
40
        "<< default_args_retries >>": 3,
41
        "<< schedule >>": "0 1 * * *",
42
        "<< bash_command_task_1 >>": "echo 'Task 1: Machine Learning'",
43
        "<< bash_command_task_2 >>": "echo 'Task 2: Machine Learning'",
44
        "<< bash_command_task_3 >>": "echo 'Task 3: Machine Learning'",
45
    },
46
]
47
48
49
populated_templates: dict = {}
50
51
# Loop through the list of template dictionaries and replace the placeholders
52
# and creating a JSON object that contains the DAG definitions
53
for variables in TEMPLATE_VARIABLES:
54
    populated_template: dict = template.copy()
55
    populated_template__str: str = json.dumps(populated_template)
56
57
    for key, value in variables.items():
58
        populated_template__str = populated_template__str.replace(key, str(value))
59
60
    populated_template = json.loads(populated_template__str)
61
    populated_template__keys = list(populated_template.keys())
62
63
    # check that each DAG is associated with only one dag_id
64
    if len(populated_template__keys) == 1:
65
        # assign the DAG id as the key in the JSON object
66
        populated_templates[populated_template__keys[0]] = populated_template[
67
            populated_template__keys[0]
68
        ]
69
    else:
70
        raise Exception("Unexpected format from populated template.")
71
72
73
# write the JSON object to a YAML file this creates one file containing
74
# the DAG factory definitions for all DAGs created from the templates
75
with open("dags/dynamic_etl.yml", "w") as ingestion__config:
76
    ingestion__config.write(yaml.dump(populated_templates))