🥇Kako pokrenuti DAG koristeći Experimental API u Airflowu

Prilikom pripreme naših obrazovnih programa povremeno se suočavamo s poteškoćama u radu s određenim alatima. I u trenutku kada se suočavamo s njima, ne postoji uvijek dovoljno dokumentacije i članaka koji bi pomogli da se riješimo tog problema.

Tako je bilo, primjerice, 2015. godine kada smo u programu "Specijalist za velike podatke" koristili Hadoop klaster sa Spark na 35 istodobnih korisnika. Kako ga pripremiti za takav slučaj korištenja koristeći YARN, nije bilo jasno. Na kraju, razradivši i prolazeći kroz sve sami, napravili smo post na Habr-u i također smo nastupili na Moscow Spark Meetup.

Povijest

Ovaj put govorimo o drugom programu – Data Engineer. Na njemu naši sudionici grade dva tipa arhitekture: lambda i kappa. I u lambda arhitekturi u okviru batch obrade koristi se Airflow za prebacivanje logova iz HDFS u ClickHouse.

Sve je generalno u redu. Neka grade svoje pipeline-ove. No, postoji jedan ali: svi naši programi su tehnološki napredni s obzirom na sam proces učenja. Za provjeru labova koristimo automatske provjerivače: sudionik treba ući u svoj račun, pritisnuti gumb "Provjeri", i nakon nekog vremena vidi neku proširenu povratnu informaciju o tome što je napravio. I upravo u tom trenutku počinjemo pristupati našem problemu.

Provjera ovog lab-a organizirana je ovako: šaljemo kontrolni paket podataka u Kafka sudionika, potom Gobblin prebacuje taj paket podataka na HDFS, zatim Airflow uzima taj paket podataka i stavlja ga u ClickHouse. Caka je u tome što Airflow to ne smije raditi u realnom vremenu, on to radi prema rasporedu: svaka 15 minuta uzima paket datoteka i prebacuje ih.

Ispada da trebamo nekako samoaktivirati njihov DAG prema našem zahtjevu tijekom rada provjeri ovdje i sada. Pretražujući, otkrili smo da za kasnije verzije Airflow postoji takozvani Experimental API. Riječ experimental, naravno, zvuči zastrašujuće, ali što učiniti… Možda će uspjeti.

Dalje opisujemo cijeli put: od instalacije Airflow-a do formiranja POST zahtjeva koji aktivira DAG koristeći Experimental API. Radit ćemo s Ubuntu 16.04.

1. Instalacija Airflow-a

Provjerit ćemo imamo li Python 3 i virtualenv.

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

Ako nešto od ovoga nedostaje, instalirajte.

Sada kreirajmo direktorij u kojem ćemo dalje raditi s Airflow-om.

$ mkdir 
$ cd /putanja/do/vašeg/novog/direktorija
$ virtualenv -p koji python3 venv
$ source venv/bin/activate
(venv) $

Instaliramo Airflow:

(venv) $ pip install airflow

Verzija s kojom smo radili: 1.10.

Sada trebamo stvoriti direktorij airflow_home, gdje će se nalaziti DAG datoteke i Airflow dodaci. Nakon stvaranja direktorija postavit ćemo varijablu okruženja AIRFLOW_HOME.

(venv) $ cd /putanja/do/mog/airflow/radnog_prostora
(venv) $ mkdir airflow_home
(venv) $ export AIRFLOW_HOME=

Sljedeći korak je izvršiti naredbu koja će stvoriti i inicijalizirati bazu podataka radnog toka u SQLite-u:

(venv) $ airflow initdb

Baza podataka će biti stvorena u airflow.db po zadanim postavkama.

Provjerimo je li Airflow instaliran:

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

Ako je naredba uspješno izvršena, Airflow je stvorio svoj konfiguracijski datoteka airflow.cfg u AIRFLOW_HOME:

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

Airflow ima web sučelje. Može se pokrenuti izvršavanjem naredbe:

(venv) $ airflow webserver --port 8081

Sada možete pristupiti web sučelju u pregledniku na portu 8081 na hostu na kojem je Airflow pokrenut, na primjer: <hostname:8081>.

2. Rad s Experimental API

Sada je Airflow postavljen i spreman za rad. Ipak, trebamo pokrenuti i Experimental API. Naši provjerači su napisani u Python-u, pa će sljedeći svi zahtjevi biti u tom jeziku koristeći biblioteku requests.

U stvarnosti, API već radi za jednostavne zahtjeve. Na primjer, ovaj zahtjev omogućava testiranje njegovog rada:

>>> import requests
>>> host = 
>>> airflow_port = 8081 #u našem slučaju ovakav, a po zadanim postavkama 8080
>>> requests.get('http://{}:{}{}'.format(host, airflow_port, '/api/experimental/test')).text
'OK'

Ako ste primili ovu poruku kao odgovor, to znači da sve radi.

Međutim, kada želimo aktivirati DAG, naići ćemo na to da ovaj tip zahtjeva ne možemo izvršiti bez autentifikacije.

Za to će biti potrebno napraviti još nekoliko koraka.

Prvo, u konfiguraciju treba dodati ovo:

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

Zatim, trebate stvoriti svog korisnika s adminsikim pravima:

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

Zatim, potrebno je stvoriti korisnika s običnim pravima, kojem će biti dozvoljeno pokretanje DAG-a.

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

Sada je sve spremno.

3. Pokretanje POST zahtjeva

Sam POST zahtjev će izgledati ovako:

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

Zahtjev je uspješno obrađen.

Stoga, nakon toga dajemo malo vremena DAG-u za obradu i radimo upit u ClickHouse tablicu, pokušavajući uhvatiti kontrolni paket podataka.

Provjera je završena.

Izvor: habr.com