🥇Ako spustiť DAG pomocou Airflow a Experimental API

Pri príprave našich vzdelávacích programov sa občas stretávame s problémami z hľadiska práce s niektorými nástrojmi. A v momente, keď sa s nimi stretávame, nie vždy je k dispozícii dostatok dokumentácie a článkov, ktoré by pomohli vyriešiť tieto problémy.

Tak to bolo napríklad v roku 2015 a v programe “Špecialista na veľké dáta” sme používali Hadoop klaster so Spark na 35 súbežných používateľov. Ako ho pripraviť na takýto use case pomocou YARN, nebolo jasné. Nakoniec, po tom, čo sme sa s tým rozobrali a prešli si cestu sami, sme vytvorili príspevok na Habr a ešte sme mali vystúpenie na Moscow Spark Meetup.

Pozadie

Tentoraz sa bude hovoriť o inom programe – Data Engineer. Na ňom naši účastníci budujú dva typy architektúry: lambda a kappa. A v lambda-architektúre v rámci spracovania dávok sa používa Airflow na presúvanie logov z HDFS do ClickHouse.

Vo všeobecnosti je všetko v poriadku. Nech si budujú svoje pipeline. Jedno však je: všetky naše programy sú technologicky náročné z pohľadu samotného procesu učenia. Na overenie laboratórií používame automatické kontroly: účastník sa musí prihlásiť do svojho osobného účtu, stlačiť tlačidlo “Overiť” a po chvíli dostane nejakú rozšírenú spätnú väzbu k tomu, čo urobil. A práve v tomto momente sa začíname približovať k našemu problému.

Overenie tejto laboratória prebieha takto: posielame kontrolný balík dát do Kafka účastníka, ďalej Gobblin presúva tento balík dát na HDFS, potom Airflow vezme tento balík dát a uloží ho do ClickHouse. Zaujímavosťou je, že Airflow to nesmie robiť v reálnom čase, robí to podľa rozvrhu: raz za 15 minút vezme balíček súborov a nahodí ich.

Tým pádom potrebujeme nejakým spôsobom sami spúšťať ich DAG na základe našich požiadaviek počas práce kontrolóra tu a teraz. Po vyhľadaní sme zistili, že pre neskoršie verzie Airflow existuje tzv. Experimental API. Slovo experimentálne, samozrejme, znie strašidelne, ale čo robiť… Možno to vyjde.

Ďalej popíšeme celú cestu: od inštalácie Airflow po vytvorenie POST požiadavky, ktorá spúšťa DAG pomocou Experimental API. Pracovať budeme s Ubuntu 16.04.

1. Inštalácia Airflow

Skontrolujeme, či máme nainštalovaný Python 3 a virtualenv.

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

Ak niečo z toho chýba, nainštalujte to.

Teraz vytvoríme adresár, v ktorom budeme ďalej pracovať s Airflow.

$ mkdir 
$ cd /cesta/k/novému/adresáru
$ virtualenv -p which python3 venv
$ source venv/bin/activate
(venv) $

Nainštalujeme Airflow:

(venv) $ pip install airflow

Verzia, na ktorej sme pracovali: 1.10.

Teraz musíme vytvoriť adresár airflow_home, kde budú umiestnené DAG súbory a pluginy Airflow. Po vytvorení adresára nastavíme premennú prostredia AIRFLOW_HOME.

(venv) $ cd /cesta/k/mojej/airflow/pracovnej_ploše
(venv) $ mkdir airflow_home
(venv) $ export AIRFLOW_HOME=

Nasledujúci krok je vykonať príkaz, ktorý vytvorí a inicializuje databázu pracovného toku v SQLite:

(venv) $ airflow initdb

Databáza bude vytvorená v airflow.db Pravidlá schvaľovania merge requestov

Skontrolujeme, či sa Airflow nainštaloval:

$ 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

Ak príkaz prebehol, Airflow vytvoril svoj konfiguračný súbor airflow.cfg v AIRFLOW_HOME:

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

Airflow má webové rozhranie. Môže sa spustiť vykonaním príkazu:

(venv) $ airflow webserver --port 8081

Teraz sa môžete dostať do webového rozhrania v prehliadači na porte 8081 na hostiteľovi, kde bol Airflow spustený, napríklad: <hostname:8081>.

2. Práca s Experimentálnym API

Teraz je Airflow nastavený a pripravený na prácu. Napriek tomu, potrebujeme spustiť ešte aj Experimentálne API. Naše kontroléry sú napísané v Pythone, takže ďalej budú všetky požiadavky na ňom s využitím knižnice requests.

V skutočnosti API už funguje pre jednoduché požiadavky. Napríklad, tento požiadavok umožňuje otestovať jeho funkciu:

>>> import requests
>>> host = 
>>> airflow_port = 8081 #v našom prípade taký, a podľa predvoleného nastavenia 8080
>>> requests.get('http://{}:{}//{}'.format(host, airflow_port, 'api/experimental/test').text
'OK'

Ak sme dostali takúto správu v odpovedi, znamená to, že všetko funguje.

Avšak, keď budeme chcieť spustiť DAG, narazíme na problém, že tento druh požiadavky nie je možné vykonať bez autentifikácie.

Na to bude potrebné vykonať ešte niekoľko akcií.

Po prvé, do konfiguračného súboru je potrebné pridať toto:

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

Potom je potrebné vytvoriť svojho používateľa s administrátorskými právami:

>>> 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()

Následne je potrebné vytvoriť používateľa so základnými právami, ktorému bude dovolené spúšťať DAG.

>>> 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()

Teraz je všetko pripravené.

3. Spustenie POST požiadavky

POST požiadavka bude vyzerať takto:

>>> 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 "n}n'

Žiadosť bola úspešne spracovaná.

Ľahko, následne dáme DAG-u nejaký čas na spracovanie a vykonáme dotaz v tabuľke ClickHouse, snažiac sa zachytiť kontrolný balík dát.

Kontrola bola dokončená.

Zdroj: habr.com