Како направити ДАГ окидач у Аирфлов-у користећи експериментални АПИ

У припреми наших образовних програма периодично наилазимо на тешкоће у раду са неким алатима. А у тренутку када их наиђемо, нема увек довољно документације и чланака који би помогли да се избори са овим проблемом.

Тако је било, на пример, 2015. године, а користили смо Хадооп кластер са Спарком за 35 истовремених корисника у програму Биг Дата Специалист. Није било јасно како га припремити за такав кориснички случај користећи ИАРН. Као резултат тога, након што су сами схватили и прошетали стазом, они су то урадили пост на Хабре а такође и извођена Мосцов Спарк Меетуп.

praistorija

Овај пут ћемо причати о другачијем програму - Дата Енгинеер. На њему наши учесници граде два типа архитектуре: ламбда и капа. А у ламдба архитектури, Аирфлов се користи као део групне обраде за пренос евиденције са ХДФС-а на ЦлицкХоусе.

Све је генерално добро. Нека граде своје цевоводе. Међутим, постоји „али“: сви наши програми су технолошки напредни у смислу самог процеса учења. Да бисмо проверили лабораторију, користимо аутоматске провере: учесник треба да оде на свој лични налог, кликне на дугме „Провери“ и након неког времена види неку врсту проширене повратне информације о томе шта је урадио. И у овом тренутку почињемо да приступамо нашем проблему.

Провера ове лабораторије је уређена на следећи начин: шаљемо пакет контролних података Кафки учесника, затим Гоблин преноси овај пакет података у ХДФС, затим Аирфлов узима овај пакет података и ставља га у ЦлицкХоусе. Трик је у томе што Аирфлов не мора то да ради у реалном времену, већ по распореду: једном сваких 15 минута узима гомилу датотека и поставља их.

Испоставило се да морамо некако сами да покренемо њихов ДАГ на наш захтев док чеккер ради овде и сада. Гуглајући, сазнали смо да за касније верзије Аирфлов-а постоји тзв Експериментални АПИ. Реч experimental, наравно, звучи застрашујуће, али шта да се ради ... Одједном се узлети.

Затим ћемо описати цео пут: од инсталирања Аирфлов-а до генерисања ПОСТ захтева који покреће ДАГ користећи експериментални АПИ. Радићемо са Убунту 16.04.

1. Инсталација протока ваздуха

Хајде да проверимо да ли имамо Питхон 3 и виртуаленв.

$ python3 --version
Python 3.6.6
$ virtualenv --version
15.2.0

Ако један од ових недостаје, инсталирајте га.

Сада направимо директоријум у коме ћемо наставити да радимо са Аирфлов-ом.

$ mkdir <your name of directory>
$ cd /path/to/your/new/directory
$ virtualenv -p which python3 venv
$ source venv/bin/activate
(venv) $

Инсталирајте проток ваздуха:

(venv) $ pip install airflow

Верзија на којој смо радили: 1.10.

Сада треба да креирамо директоријум airflow_home, где ће се налазити ДАГ датотеке и додаци Аирфлов. Након креирања директоријума, подесите променљиву окружења AIRFLOW_HOME.

(venv) $ cd /path/to/my/airflow/workspace
(venv) $ mkdir airflow_home
(venv) $ export AIRFLOW_HOME=<path to airflow_home>

Следећи корак је да покренете команду која ће креирати и иницијализовати базу података тока података у СКЛите-у:

(venv) $ airflow initdb

База података ће бити креирана у airflow.db Уобичајено.

Проверите да ли је Аирфлов инсталиран:

$ airflow version
[2018-11-26 19:38:19,607] {__init__.py:57} INFO - Using executor SequentialExecutor
[2018-11-26 19:38:19,745] {driver.py:123} INFO - Generating grammar tables from /usr/lib/python3.6/lib2to3/Grammar.txt
[2018-11-26 19:38:19,771] {driver.py:123} INFO - Generating grammar tables from /usr/lib/python3.6/lib2to3/PatternGrammar.txt
  ____________       _____________
 ____    |__( )_________  __/__  /________      __
____  /| |_  /__  ___/_  /_ __  /_  __ _ | /| / /
___  ___ |  / _  /   _  __/ _  / / /_/ /_ |/ |/ /
 _/_/  |_/_/  /_/    /_/    /_/  ____/____/|__/
   v1.10.0

Ако је команда функционисала, онда је Аирфлов креирао сопствену конфигурациону датотеку airflow.cfg в AIRFLOW_HOME:

$ tree
.
├── airflow.cfg
└── unittests.cfg

Аирфлов има веб интерфејс. Може се покренути покретањем команде:

(venv) $ airflow webserver --port 8081

Сада можете да приступите веб интерфејсу у претраживачу на порту 8081 на хосту где је Аирфлов био покренут, овако: <hostname:8081>.

2. Рад са експерименталним АПИ-јем

На овом Аирфлов је конфигурисан и спреман за рад. Међутим, такође морамо да покренемо експериментални АПИ. Наши чекери су написани на Питхон-у, тако да ће даље сви захтеви бити на њему користећи библиотеку requests.

У ствари, АПИ већ ради за једноставне захтеве. На пример, такав захтев вам омогућава да тестирате његов рад:

>>> import requests
>>> host = <your hostname>
>>> airflow_port = 8081 #в нашем случае такой, а по дефолту 8080
>>> requests.get('http://{}:{}/{}'.format(host, airflow_port, 'api/experimental/test').text
'OK'

Ако сте добили такву поруку као одговор, то значи да све функционише.

Међутим, када желимо да покренемо ДАГ, наилазимо на чињеницу да се ова врста захтева не може упутити без аутентификације.

Да бисте то урадили, мораћете да извршите низ радњи.

Прво, морате додати ово у конфигурацију:

[api]
auth_backend = airflow.contrib.auth.backends.password_auth

Затим морате да креирате свог корисника са администраторским правима:

>>> import airflow
>>> from airflow import models, settings
>>> from airflow.contrib.auth.backends.password_auth import PasswordUser
>>> user = PasswordUser(models.Admin())
>>> user.username = 'new_user_name'
>>> user.password = 'set_the_password'
>>> session = settings.Session()
>>> session.add(user)
>>> session.commit()
>>> session.close()
>>> exit()

Затим морате да креирате корисника са нормалним правима коме ће бити дозвољено да направи ДАГ окидач.

>>> import airflow
>>> from airflow import models, settings
>>> from airflow.contrib.auth.backends.password_auth import PasswordUser
>>> user = PasswordUser(models.User())
>>> user.username = 'newprolab'
>>> user.password = 'Newprolab2019!'
>>> session = settings.Session()
>>> session.add(user)
>>> session.commit()
>>> session.close()
>>> exit()

Сада је све спремно.

3. Покретање ПОСТ захтева

Сам ПОСТ захтев ће изгледати овако:

>>> dag_id = newprolab
>>> url = 'http://{}:{}/{}/{}/{}'.format(host, airflow_port, 'api/experimental/dags', dag_id, 'dag_runs')
>>> data = {"conf":"{"key":"value"}"}
>>> headers = {'Content-type': 'application/json'}
>>> auth = ('newprolab', 'Newprolab2019!')
>>> uri = requests.post(url, data=json.dumps(data), headers=headers, auth=auth)
>>> uri.text
'{n  "message": "Created <DagRun newprolab @ 2019-03-27 10:24:25+00:00: manual__2019-03-27T10:24:25+00:00, externally triggered: True>"n}n'

Захтев је успешно обрађен.

Сходно томе, онда дајемо ДАГ-у мало времена да обради и упути захтев табели ЦлицкХоусе, покушавајући да ухвати контролни пакет података.

Верификација је завршена.

Извор: ввв.хабр.цом