TL; DR;
Der Artikel beschreibt, wie Apache Airflow implementiert wurde, um auf QlikView basierende Berichtsaktualisierungsjobs in einer relativ umfangreichen Implementierung zu verwalten.
Die AbhĂ€ngigkeitshölle (englische AbhĂ€ngigkeitshölle) ist das Gegenmuster des Konfigurationsmanagements, die Erweiterung des Diagramms der gegenseitigen AbhĂ€ngigkeiten von Softwareprodukten und Bibliotheken, was zu der Schwierigkeit fĂŒhrt, neue Produkte zu installieren und alte zu deinstallieren.
Wikipedia
, 2018 , BIA-Technologies.
QlikView " ".
Qlik QlikSense, QlikView, 12.4, .
, , .
QlikView "" (application) â , "QVW" ( ), , , .
, .
-- , , - QV. . , , . , "", ETL- , .
QV QVW , : , , . .
, , QVD. QlikView , , .
, ETL- , , QVD, , .
Qlik , , , (), - . QlikView QVD- . , QlikView, , , , .
, Qlik â NPrinting, , , , , SSRS, , QlikView, .
, QlikView:
- , , QlikView Server (QVS). .QVW .
- QlikView Distribution Service (QDS). Publisher , Reload Engine, QlikView, : , QVS, " ", , , , . , , , QlikView, , .
- , QlikView Management Service (QMS) â .
- , , - â
, MS SQL, ETL QlikView .
( ) , .QVW , . , . , , , QVD.
:
- , , .
- , - , 8-9 , . , , - , - 4 , , .
QMS , , , . , ( Reload Engine, . Publisher , rusbaron , ).
QMS :
, , .
, QMS . (, ).
, , , .
, .
, .
, .
, , , ( ), , , - , ..
, .
- , , , . , , .
, , ETL-, , , . , , , ETL- .
.
, â , , ,- â , .
- , , .
, . - , -. .
, . , , , . , , , . , , â .
, QDS . , - .
, . : , .
? . . ? .
, QlikView , , , .
, , , QlikView, . , , Hadoop ETL- .
QlikView , . , , .
QlikView : QMS QlikView Management Service API, . EDX (Event Driven Execution). , , QMS .
, " ", , , , , NPrinting.
, , â (Workflow management systems). , Apache Airflow.
Airflow â , . , , :
â , -.
Airflow , , Airflow QMS, Airflow â , , . , , : -> Airflow â QlikView.
API QMS . , .
Airflow â , . , , , , .
Airflow . , (, , ), QlikView, , , .
Airflow . , , - . , , : , . , , , , , , .
Airflow (DAG â directed acyclic graph), , â .
, "Application 7" "ETL_4" "ETL_5". "ETL_3" "TimeSensor_6_30" â , 6:30.
Python.
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',
]},
}
Airflow Python, , , . QlikView . , ( , , ) DAG. , , DAG .
: (5 ), (1 ) (100 ). . Airflow â â , , , , , , , , .
, DAG,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()
, Airflow QlikView:
Airflow .
.
- , - , .
, ETL- , , .
, Airflow . .
, , .
, â 1 . , -, .
. , , , . , .
.
Airflow erstellt sehr nĂŒtzliche Berichte zu Aufgabenleistungsstatistiken. Wir können die Wachstumsdynamik der Aufgabendauer verfolgen und andere Berichte erhalten, z. B. ein Gantt-Diagramm:
Combat Airflow arbeitet mit uns in einer virtuellen Maschine mit zwei Kernen und 4 GB RAM. Dies reicht fĂŒr den reibungslosen Betrieb des Dienstes aus.
Im Allgemeinen glaube ich, dass wir das Problem der Aktualisierung der QlikView-Berichterstellung erfolgreich gelöst haben, und Airflow hat uns dabei geholfen, von unschÀtzbarem Wert zu sein.
