example_pipeline.py

"""
### My Example DAG using Taskflow API

This document will show on the Tree View, Graph View, and Calendar View.
"""
from airflow.decorators import dag, task
from airflow.operators.dummy import DummyOperator
from airflow.utils import timezone


EMAILS = ["[email protected]",]
default_args = {
    "owner": "zkan",
    "email": EMAILS,
    "start_date": timezone.datetime(2021, 6, 9),
}
@dag(default_args=default_args, schedule_interval="@daily", catchup=False, tags=["dataength"])
def example_pipeline():

    start = DummyOperator(task_id="start")
    start.doc_md = """
    This task is a starting point. It will show on the Task Instance Details.
    """

    @task
    def sum():
        return 1 + 1

    @task
    def multiply_by_ten(total):
        return total * 10


    @task
    def double(total):
        return total * 2


    sum_result = sum()
    sum_result.doc_md = """
    Compute the sum of 1 and 1.
    """

    multiply_result = multiply_by_ten(sum_result)
    multiply_result.doc_md = """
    Compute the multiplication of the sum result
    from the previous task and 10.
    """

    double_result = double(multiply_result)
    double_result.doc_md = """
    Compute the double of the multiplication result
    from the previous task.
    """

    end = DummyOperator(task_id="end")
    end.doc_md = """
    This task is an ending point.
    """

    start >> sum_result
    double_result >> end


dag = example_pipeline()
dag.doc_md = __doc__