الخروج من جحيم التبعية في QlikView

, , , QlikView, . , , Hadoop ETL- .

from datetime import datetime, timedelta
from airflow import DAG
from airflow.sensors.time_delta_sensor import TimeDeltaSensor
from airflow.contrib.operators.qds_operator import QDSReloadApplicationOperator

default_args = {
    'owner': 'airflow',
    'depends_on_past': False,
    'start_date': datetime(2018, 1, 19),
    'email': ['Airflow.Administrator@mycompany.ru'],
    'email_on_failure': True,
    'email_on_retry': False,
    'retries': 0,
    'retry_delay': timedelta(minutes=5),
    'pool': 'default_pool',
}

dag = DAG('example_qds_operator',
          description='Reload applications at QlikView Distribution Service',
          catchup=False,
          schedule_interval='0 21 * * 5',
          default_args=default_args)

tasksDict = {
    u'ETL_1.qvw': {},

    u'ETL_2.qvw': {
        'Pool': 'Heavy_ETL_pool',
    },

    u'ETL_3.qvw': {
        'StartTime': [6, 30]},

    u'ETL_4.qvw': {
        'Priority': -5,
        'Dep': [
            u'ETL_1.qvw',
            u'ETL_3.qvw', ]},

    u'ETL_5.qvw': {
        'Dep': [
            u'ETL_2.qvw', ]},

    u'Application_6.qvw': {
        'Dep': [
            u'ETL_1.qvw',
            u'ETL_5.qvw', ]},

    u'Application_7.qvw': {
        'Dep': [
            u'ETL_4.qvw',
            u'ETL_5.qvw',
        ]},
}

airflowTasksDict = {}

for task in tasksDict.keys():
    task_id = task.replace(" ", "_").replace("'", "").replace("/", "_").replace("(", "_").replace(")", "_").replace(",", "_").replace(".qvw", "").replace("__",
                                                                                                                                                          "_")
    AirflowTask = QDSReloadApplicationOperator(document_name=task, task_id=task_id, qv_conn_id='qv_connection', dag=dag)
    airflowTasksDict[task] = AirflowTask

for task in tasksDict.keys():
    if 'Dep' in tasksDict[task]:
        for dep in tasksDict[task]['Dep']:
            airflowTasksDict[task].set_upstream(airflowTasksDict[dep])

    if 'Pool' in tasksDict[task]:
        airflowTasksDict[task].pool = tasksDict[task]['Pool']

    if 'Priority' in tasksDict[task]:
        airflowTasksDict[task].priority_weight = tasksDict[task]['Priority']

    if 'StartTime' in tasksDict[task]:
        hour = tasksDict[task]['StartTime'][0]
        minute = tasksDict[task]['StartTime'][1]
        sensorTime = timedelta(hours=hour, minutes=minute)
        sensorTaskID = u'TimeSensor_{}_{}'.format(hour, minute)

        if sensorTaskID not in airflowTasksDict:
            SensorTask = TimeDeltaSensor(delta=sensorTime, task_id=sensorTaskID, pool='Sensors', dag=dag)
            airflowTasksDict[sensorTaskID] = SensorTask
        airflowTasksDict[task].set_upstream(airflowTasksDict[sensorTaskID])

if __name__ == '__main__':
    dag.clear(reset_dag_runs=True)
    dag.run()