• Python
• Анализ и проектирование систем
• Визуализация данных
• Софт
• Data Engineering

TL;DR;

В статье описано, как внедрялся Apache Airflow для управления заданиями обновления отчетности, построенной на QlikView в достаточно крупном внедрении.

Dependency hell (англ. ад зависимостей) — это антипаттерн управления конфигурацией, разрастание графа взаимных зависимостей программных продуктов и библиотек, приводящее к сложности установки новых и удаления старых продуктов.

Википедия

Введение

Речь пойдет о событиях, которые происходили весной 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 есть возможность задавать расписание обновления и зависимости файлов, но, к сожалению, этот функционал достаточно ограничен. Например, мы можем задать зависимость только от одного файла.

Выглядит интерфейс к QMS примерно так:

В нашем же случае количество зависимостей может быть достаточно большим, так как справочники и таблицы фактов для итогового отчета могут готовиться более, чем десятком других приложений.

Также, в QMS у нас нет возможности задавать приоритеты обновления отчетности. Нет возможности настраивать штатным образом запуск обновления отчетности по внешнему событию (например, по факту появления данных в источнике).

Долгое время мы боролись с этими проблемами, вручную разнося обновляемые приложения по разным веткам, и задавая разные расписания, пытаясь таким образом раздвинуть зависимые отчеты друг от друга во времени.

Когда приложений в нашей системы было немного, это отлично работало.
Но потом в процессе эксплуатации их количество перевалило за несколько десятков, мы начали сталкиваться с проблемами.

• Процесс обновления отчетности растянулся на все доступное нам окно, и норовил вылезти дальше.

  
  

  Дело в том, что продолжительность выполнения одной задачи непостоянное, оно зависит от количества данных, которые обрабатываются (а оно постоянно растет), от того, как распределяются ресурсы сервера между параллельно запущенными процессами обновления, от загруженности сервиса-источника данных, и т.д.

  
  

  То есть длительность обновления конкретного приложения плавает, и иногда значительно.

  
  

  И из-за того, что нам сложно было угадать, когда же в действительно закончится обновление всех зависимостей, нам приходилось задавать большие интервалы свободного времени перед запусками обновлений зависимых приложений. Чем длиннее цепочка приложений, тем больше интервалов между ними, тем дольше длится весь процесс.

  
  

  Кроме того, чтобы снизить хаос и упростить себе процесс администрирования системы, часть ETL-процессов, которые зависели от одних источников, мы объединяли в один скрипт, в котором эти процессы выполнялись последовательно. Это уменьшает вероятность того, что данные будут обновлены не в том порядке, но удлиняет общее время выполнения цикла обновления, потому что мы лишаемся возможности часть из этих ETL-процедур выполнять параллельно.

  
  

• Участились падения обновления отчетности.

  
  

  Кроме естественных причин, по которым может упасть обновление — недоступность источника данных, обрыв коннекта, ошибка в скрипте,- падение может произойти по причине одновременного доступа к файлу из нескольких задач — попытка обновить файл, читаемый в этот момент другим процессом.

  
  

  И из-за того, что невозможно было в штатном планировщике учесть все зависимости для большого количества файлов, все чаще и чаще происходили такие моменты.

  
  

• Участились инциденты с проблемой необновленных данных

  
  

  Все чаще и чаще происходили ситуации, когда задачи выполнялись в неправильном порядке. И тогда приложение-потребитель брало данные из файла, который еще не был обновлен приложением-поставщиком. В итоге конечный пользователь видел в отчете неактуальную информацию.

  
  

• Падение одного отчета стопорило процесс обновления большого количества других отчетов

  
  

  Кроме задания максимального количества одновременно выполняющихся заданий, мы выстраивали иерархии обновлений из приложений в том числе и для того, чтобы задать порядок обновления отчетов. Ресурсы сервера не бесконечные, и, например, часть приложений, которые оказались весьма тяжелыми, перелопачивают данные в объеме порядка миллиарда строк, занимая почти всю доступную на сервере память. Допустить одновременное выполнение нескольких таких заданий нельзя, поэтому они выстраивались в иерархии одно за другим. И получается, что логически задания не должны зависеть друг от друга, но физически — зависят.

  
  

  В случае падения обновления одного отчета, QDS отменяет обновление всех подветок зависимых отчетов. А для нас было бы полезно, если б часть из тех приложений все-таки выполнилось.

  
  

• Администрирование всей системы стало адом зависимостей

  
  

  Хрупкое равновесие процесса обновления всей грозди отчетов, достигнутое бесчисленным количеством проб и ошибок мгновенно рассыпалось при необходимости добавить в систему новый отчет. И каждая новая итерация обновления отчетности таила в себе интригу: что же отвалится на этот раз, и почему.

  
  

  Отчет упал? Сдвигаем его в расписании на другое время. Или ставим в зависимость от другого отчета. На следующий день снова проблема? Опять начинаем манипуляции.

  
  

В итоге, когда в нашей системе отчетности QlikView стало под сотню приложений, мы поняли, что дальше так продолжаться не может, и начали искать системное решение этой проблемы.

Что мы сделали

Раз родной сервис от 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 — сенсоры — такие же задачи, что и остальные, поэтому если для них не выделить отдельный пул, то они могут, например, или занять весь пул задач, или наоборот, ожидая своего запуска в очереди, не успеть своевременно выполниться.

Что имеем в итоге

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 строит достаточно полезную отчетность по статистике выполнения задач, и мы можем, отслеживать динамику роста продолжительности заданий, а также получать другие отчеты, например, диаграмму Ганта:

Боевой Airflow работает у нас на виртуальной машине с двумя ядрами и 4 Gb ОЗУ, и этого вполне хватает для бесперебойной работы сервиса.

В целом, я считаю, что мы успешно справились с решением проблемы с обновлением отчетности QlikView, и Airflow неоценимо нам в этом помог.