🥇Kako aktivirati DAG v Airflowu z uporabo Experimental API

Med pripravo naših izobraževalnih programov se občasno srečujemo s težavami pri delu z nekaterimi orodji. In v tistem trenutku, ko naletimo nanje, pogosto ni dovolj dokumentacije in člankov, ki bi nam pomagali rešiti problem.

Tako je bilo, na primer, leta 2015 in na programu 'Specialist za velike podatke' smo uporabljali Hadoop grozd s Spark za 35 hkratnih uporabnikov. Kako ga pripraviti za takšno uporabo z YARN, ni bilo jasno. Na koncu smo, ko smo se sami prebijali skozi to, ustvarili prispevek na Habré in še nastopili na Moscow Spark Meetup.

Pozadina

Tokrat bo govora o drugem programu – Data Engineer. Na njem naši udeleženci gradijo dva tipa arhitekture: lambda in kappa. V lambda arhitekturi se v okviru serijske obdelave uporablja Airflow za prenos logov iz HDFS v ClickHouse.

Vse je na splošno dobro. Naj gradijo svoje cevi. Vendar pa obstaja ena težava: vsi naši programi so tehnološki z vidika samega procesa učenja. Za preverjanje laboratorijskih del uporabljamo avtomatske preverjače: udeleženec se mora prijaviti v svoj račun, pritisniti gumb 'Preveri' in čez nekaj časa vidi neko razširjeno povratno informacijo o tem, kar je naredil. In prav v tem trenutku se začnemo ukvarjati z našim problemom.

Ocenjevanje te laboratorijske naloge je zasnovano tako: pošljemo kontrolni paket podatkov v Kafka udeleženca, nato Gobblin prenese ta paket podatkov na HDFS, nato Airflow vzame ta paket podatkov in ga položi v ClickHouse. Ključna točka je, da Airflow tega ne sme delati v realnem času, temveč to počne po urniku: vsake 15 minut vzame paket datotek in jih naloži.

Izkazalo se je, da moramo nekako sprožiti njihov DAG sami na zahtevo med delom preverjača tukaj in zdaj. Po Googlanju smo ugotovili, da za novejše različice Airflow obstaja tako imenovani Eksperimentalni API. Beseda eksperimentalni, seveda, zveni strašljivo, toda kaj storiti… Morda bo uspelo.

Nato bomo opisali celoten postopek: od namestitve Airflow do oblikovanja POST zahteve, ki sproži DAG, z uporabo Eksperimentalnega API. Delali bomo z Ubuntu 16.04.

1. Namestitev Airflow

Preverili bomo, da imamo Python 3 in virtualenv.

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

Če česar od tega ni, ga namestite.

Sedaj bomo ustvarili mapo, v kateri bomo nadalje delali z Airflow.

$ mkdir 
$ cd /pot/do/vašega/ novega/direktorija
$ virtualenv -p which python3 venv
$ source venv/bin/activate
(venv) $

Namestimo Airflow:

(venv) $ pip install airflow

Različica, s katero smo delali: 1.10.

Zdaj moramo ustvariti imenik airflow_home, kjer bodo shranjeni DAG datoteke in vtičniki Airflow. Po ustvaritvi imenika bomo nastavili okoljsko spremenljivko AIRFLOW_HOME.

(venv) $ cd /pot/do/moje/airflow/ponudbe
(venv) $ mkdir airflow_home
(venv) $ export AIRFLOW_HOME=

Naslednji korak je, da izvedemo ukaz, ki bo ustvaril in inicializiral podatkovno bazo potoka podatkov v SQLite:

(venv) $ airflow initdb

Podatkovna baza bo ustvarjena v airflow.db po defaultu.

Preverimo, ali je Airflow pravilno nameščen:

$ airflow version
[2018-11-26 19:38:19,607] {__init__.py:57} INFO - Uporaba izvršitelja SequentialExecutor
[2018-11-26 19:38:19,745] {driver.py:123} INFO - Generiranje tabel slovnice iz /usr/lib/python3.6/lib2to3/Grammar.txt
[2018-11-26 19:38:19,771] {driver.py:123} INFO - Generiranje tabel slovnice iz /usr/lib/python3.6/lib2to3/PatternGrammar.txt
  ____________       _____________
 ____    |__( )_________  __/__/  /________      __
____  /| |_  /__  ___/_  /_ __  /_  __ _ | /| /  /
___  ___ |  / _  /   _  __/ _  / /  /_/_  |/ |/  /
 _/_/  |_/_/  /_/_/    /_/    /_/  ____/____/_|__/
   v1.10.0

Če je ukaz deloval, je Airflow ustvaril svoj konfiguracijski datoteko airflow.cfg u AIRFLOW_HOME:

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

Airflow ima spletno vmesnik. Zažene se ga lahko z naslednjim ukazom:

(venv) $ airflow webserver --port 8081

Sedaj lahko dostopate do spletnega vmesnika v brskalniku na portu 8081 na gostitelju, kjer je bil Airflow zagnan, na primer: <hostname:8081>.

2. Delo z Experimental API

Zdaj je Airflow nastavljen in pripravljen za delo. Vendar moramo zagnati tudi Experimental API. Naši preverjevalniki so napisani v Pythonu, zato bodo naslednji vsi zahtevki napisani v njem z uporabo knjižnice requests.

Pravzaprav API že deluje za preproste zahteve. Na primer, ta zahteva omogoča preizkušanje njegovega delovanja:

>>> import requests
>>> host = 
>>> airflow_port = 8081 #v našem primeru je tak, privzeto pa 8080
>>> requests.get('http://{}:{} /{}'.format(host, airflow_port, 'api/experimental/test')).text
'OK'

Če ste prejeli to sporočilo kot odgovor, to pomeni, da vse deluje.

Vendar pa, ko bomo želeli aktivirati DAG, se bomo soočili s težavo, da tega tipa zahtevo ni mogoče izvesti brez avtentikacije.

Za to bo potrebno izvesti še vrsto dejanj.

Prvič, v konfiguracijo je treba dodati tole:

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

Nato je potrebno ustvariti svojega uporabnika z administrativnimi pravicami:

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

Potem je treba ustvariti uporabnika s standardnimi pravicami, ki mu bo dovoljeno aktivirati 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()

Zdaj je vse pripravljeno.

3. Zagon POST zahtevka

POST zahtevek bo izgledal takole:

>>> 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'

Zahtevek je bil uspešno obdelan.

Nato damo DAG-u nekaj časa za obdelavo in naredimo zahtevek v ClickHouse tabelo, poskušajoč ujeti kontrolni paket podatkov.

Preverjanje je zaključeno.

Izvor: habr.com