How to find run-time for a set of tasks in an Airflow DAG - Stack Overflow

admin管理员组

文章数量:1244270

I have a pretty simple DAG which has some primary operational tasks, and then some cleanup and reporting tasks. I would like to get the runtime of the operational bits either by the difference of the start time of the first task and the end time of the last operational one, or if I can do this by putting them in a task group, that would be even easier.

I'm not clear how to do this though. My tasks are built on custom operators built from the base Airflow operator class.

My ideal would be

def get_time_delta(start_task, end_task) -> int:
    """Determine time delta in seconds between two airflow tasks."""
    # This is the bit I don't know where to find
    return end_task.dt - start_task.dt

@dag(...)
def my_dag():
    task1 = MyCustomOperator1(...)
    task2 = MyCustomOperator2(...)
    task3 = MyCustomEmitMetricOperator(runtime=get_time_delta(task1, task2))

    task1 >> task2 >> task3

Any support on how to achieve this would be greatly appreciated.

本文标签： How to find runtime for a set of tasks in an Airflow DAGStack Overflow