long_task_dag

from airflow import DAG
from airflow.operators.python_operator import PythonOperator
from datetime import datetime, timedelta

import time

schedule_interval = '45 7 * * *'
default_args = {
    'owner': 'user',
    'start_date': datetime(2017, 1, 13, 7, 30),
    'email': ['[email protected]']
}

dag = DAG('long_task',
          schedule_interval = schedule_interval,
          default_args=default_args)

def run(n):
    time.sleep(n)

task1 = PythonOperator(task_id='task1',
                       execution_timeout=timedelta(hours=2),
                       python_callable=run,
                       op_kwargs={'n': 5400},
                       dag=dag)
task2 = PythonOperator(task_id='task2',
                       execution_timeout=timedelta(hours=2),
                       python_callable=run,
                       op_kwargs={'n': 1800},
                       dag=dag)

task1 >> task2