Při přípravě našich vzdělávacích programů se pravidelně setkáváme s obtížemi z hlediska práce s některými nástroji. A v okamžiku, kdy se s nimi potýkáme, není vždy dostatek dokumentace a článků, které by pomohly tuto problém vyřešit.
Tak to bylo například v roce 2015, kdy jsme v programu "Specialista na velká data" používali Hadoop cluster se Spark na 35 současných uživatelů. Jak ho připravit pro takový uživatelský případ pomocí YARN, bylo nejasné. Nakonec, když jsme to pochopili a prošli si tím sami, vytvořili jsme a také jsme vystoupili na .
Pozadí
Tentokrát se budeme bavit o jiném programu – . Na něm naši účastníci staví dva typy architektury: lambda a kappa. A v lambda architektuře v rámci dávkového zpracování se používá Airflow pro převod logů z HDFS do ClickHouse.
Všechno je celkem dobré. Ať si staví své pipeline. Nicméně, je tu jedno, co nás trápí: všechny naše programy jsou technologické z hlediska samotného procesu učení. Pro kontrolu laboratoří používáme automatické checkery: účastník se musí přihlásit do osobního účtu, stisknout tlačítko „Zkontrolovat“ a po chvíli uvidí nějakou podrobnou zpětnou vazbu na to, co udělal. A právě v tuto chvíli začínáme přistupovat k našemu problému.
Kontrola této laboratoře je uspořádaná takto: posíláme kontrolní balíček dat do Kafka účastníka, poté Gobblin převádí tento balíček dat na HDFS, potom Airflow vezme tento balíček dat a uloží ho do ClickHouse. Hlavní věc je, že Airflow to nesmí dělat v reálném čase, dělá to podle plánu: každých 15 minut vezme balík souborů a zpracovává je.
To znamená, že potřebujeme nějak aktivovat jejich DAG samostatně na požadavek během práce checkera tady a teď. Po vyhledání na Googlu jsme zjistili, že pro pozdní verze Airflow existuje takzvané . Slovo experimentální, samozřejmě, zní děsivě, ale co dělat… Možná to vyjde.
Dále popíšeme celou cestu: od instalace Airflow po vytvoření POST požadavku, který aktivuje DAG, pomocí Experimental API. Pracovat budeme s Ubuntu 16.04.
1. Instalace Airflow
Zkontrolujeme, že máme nainstalovaný Python 3 a virtualenv.
$ python3 --version
Python 3.6.6
$ virtualenv --version
15.2.0Pokud něco z toho chybí, nainstalujte to.
Nyní vytvoříme adresář, ve kterém budeme dále pracovat s Airflow.
$ mkdir
$ cd /cesta/k/vaší/nové/složce
$ virtualenv -p which python3 venv
$ source venv/bin/activate
(venv) $Nainstalujeme Airflow:
(venv) $ pip install airflowVerze, na které jsme pracovali: 1.10.
Nyní potřebujeme vytvořit adresář airflow_home, kde budou umístěny DAG soubory a pluginy Airflow. Po vytvoření složky nastavíme proměnnou prostředí AIRFLOW_HOME.
(venv) $ cd /cesta/k/mému/airflow/pracovišti
(venv) $ mkdir airflow_home
(venv) $ export AIRFLOW_HOME=Dalším krokem je provést příkaz, který vytvoří a inicializuje databázi pro pracovní tok v SQLite:
(venv) $ airflow initdbDatabáze bude vytvořena v airflow.db jako výchozí.
Zkontrolujeme, zda se Airflow nainstaloval:
$ airflow version
[2018-11-26 19:38:19,607] {__init__.py:57} INFO - Používám executor SequentialExecutor
[2018-11-26 19:38:19,745] {driver.py:123} INFO - Generování gramatických tabulek z /usr/lib/python3.6/lib2to3/Grammar.txt
[2018-11-26 19:38:19,771] {driver.py:123} INFO - Generování gramatických tabulek z /usr/lib/python3.6/lib2to3/PatternGrammar.txt
____________ _____________
____ |__( )_________ __/ __ /________ __
____ /| |_ /__ ___/_ /_ __ /_ __ _ | /| / \ /
___ ___ | / _ / _ __/ _ / / /_/ __ _ |/ |/
_/_/_/ |_/_/_/ /_/ /_/ /_/ ____/____/|__/
v1.10.0Pokud příkaz proběhl úspěšně, Airflow vytvořil svůj konfigurační soubor airflow.cfg v AIRFLOW_HOME:
$ tree
.
├── airflow.cfg
└── unittests.cfgAirflow má webové rozhraní. Můžete ho spustit příkazem:
(venv) $ airflow webserver --port 8081Nyní můžete přistupovat k webovému rozhraní v prohlížeči na portu 8081 na hostiteli, kde byl Airflow spuštěn, například: <hostname:8081>.
2. Práce s Experimental API
Tímto je Airflow nastaven a připraven k použití. Musíme však spustit také Experimental API. Naši kontrolery jsou napsány v Pythonu, takže všechny další dotazy budou v něm pomocí knihovny requests.
Ve skutečnosti API již funguje pro jednoduché dotazy. Například takový dotaz umožňuje otestovat jeho funkčnost:
>>> import requests
>>> host =
>>> airflow_port = 8081 # v našem případě, ve výchozím stavu 8080
>>> requests.get('http://{}:{}{}'.format(host, airflow_port, '/api/experimental/test')).text
'OK'Pokud jste dostali takovou zprávu jako odpověď, znamená to, že vše funguje.
Když se však pokusíme spustit DAG, narazíme na to, že tento typ dotazu nelze provést bez autentizace.
K tomu bude potřeba provést ještě několik kroků.
Za prvé, do konfigurace je třeba přidat toto:
[api]
auth_backend = airflow.contrib.auth.backends.password_authPoté je třeba vytvořit svého uživatele s administrátorskými právy:
> 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()Poté je nutné vytvořit uživatele s běžnými právy, kterému bude povoleno spouštět 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()Teď je vše připraveno.
3. Spuštění POST požadavku
Samotný POST požadavek bude vypadat 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": "Vytvořeno "n}n'Požadavek byl úspěšně zpracován.
Poté dáme DAGu nějaký čas na zpracování a provedeme dotaz na tabulku ClickHouse, abychom se pokusili zachytit kontrolní datový balíček.
Kontrola byla dokončena.
Zdroj: habr.com
