airflow_dag_tutorial_with_bash_operators_and_task_dependencies.py

1import textwrap
2from datetime import datetime, timedelta
3
4# The DAG object; we'll need this to instantiate a DAG
5from airflow.models.dag import DAG
6
7# Operators; we need this to operate!
8from airflow.operators.bash import BashOperator
9
10with DAG(
11    "tutorial",
12    # These args will get passed on to each operator
13    # You can override them on a per-task basis during operator initialization
14    default_args={
15        "depends_on_past": False,
16        "email": ["airflow@example.com"],
17        "email_on_failure": False,
18        "email_on_retry": False,
19        "retries": 1,
20        "retry_delay": timedelta(minutes=5),
21    },
22    description="A simple tutorial DAG",
23    schedule=timedelta(days=1),
24    start_date=datetime(2021, 1, 1),
25    catchup=False,
26    tags=["example"],
27) as dag:
28
29    # t1, t2 and t3 are examples of tasks created by instantiating operators
30    t1 = BashOperator(
31        task_id="print_date",
32        bash_command="date",
33    )
34
35    t2 = BashOperator(
36        task_id="sleep",
37        depends_on_past=False,
38        bash_command="sleep 5",
39        retries=3,
40    )
41
42    t1.doc_md = textwrap.dedent(
43        """\
44    #### Task Documentation
45    You can document your task using the attributes `doc_md` (markdown),
46    `doc` (plain text), `doc_rst`, `doc_json`, `doc_yaml` which gets
47    rendered in the UI's Task Instance Details page.
48    ![img](https://imgs.xkcd.com/comics/fixing_problems.png)
49    **Image Credit:** Apache Airflow
50    """
51    )
52
53    dag.doc_md = __doc__  # providing that you have a docstring at the beginning of the DAG; OR
54    dag.doc_md = """
55    This is a documentation placed anywhere
56    """  # otherwise, type it like this
57    
58    templated_command = textwrap.dedent(
59        """
60    {% for i in range(5) %}
61        echo "{{ ds }}"
62        echo "{{ macros.ds_add(ds, 7) }}"
63    {% endfor %}
64    """
65    )
66
67    t3 = BashOperator(
68        task_id="templated",
69        depends_on_past=False,
70        bash_command=templated_command,
71    )
72
73    t1 >> [t2, t3]