🥇Kā izveidot DAG trigeri Airflow, izmantojot Experimental API

Saglabājot mūsu izglītības programmas, periodiski saskaramies ar grūtībām attiecībā uz darbu ar dažiem rīkiem. Un tajā brīdī, kad saskaramies ar tiem, ne vienmēr ir pietiekami daudz dokumentācijas un rakstu, kas palīdzētu risināt šo problēmu.

Tā notika, piemēram, 2015. gadā, un mūsu programma "Speciālists lielo datu jomā" izmantoja Hadoop klasteri ar Spark, kas atbalstīja 35 vienlaicīgus lietotājus. Kā to sagatavot šādam lietošanas gadījumam ar YARN, nebija skaidrs. Rezultātā, izpētot un iziet paši šo ceļu, mēs izveidojām ierakstu Habrā un vēl piedalījāmies Moscow Spark Meetup.

Iepriekšējā pieredze

Šoreiz runa būs par citu programmu – Datu inženieris. Tajā mūsu dalībnieki veido divu veidu arhitektūru: lambda un kappa. Un lambda arhitektūrā partijas apstrādes ietvaros tiek izmantots Airflow, lai pārvietotu žurnālus no HDFS uz ClickHouse.

Viss kopumā ir labi. Lai viņi būvē savus caurlietošanas gadījumus. Tomēr ir viena problēma: visas mūsu programmas ir tehnoloģiskas pašas mācību procesa dēļ. Laboratoriju pārbaudei mēs izmantojam automātiskus pārbaudītājus: dalībniekam jāpieslēdzas savam kontam, jānospiež poga "Pārbaudīt", un pēc noteikta laika viņš redz kādu paplašinātu atsauksmi par to, ko ir izdarījis. Tieši šajā brīdī mēs sākam pievērsties mūsu problēmai.

Šīs laboratorijas pārbaude ir organizēta šādi: mēs sūtām kontrolpaketi datu uz dalībnieka Kafka, pēc tam Gobblin pārvieto šo datu paketi uz HDFS, pēc tam Airflow ņem šo datu paketi un ievieto to ClickHouse. Īpašā iezīme ir tāda, ka Airflow to nedara reālā laikā, bet gan pēc grafika: ik pēc 15 minūtēm ņem failu paciņu un ievieto.

Tādējādi mums ir nepieciešams kādā veidā trigerēt viņu DAG paši pēc mūsu pieprasījuma pārbaudītāja darbības laikā. Pēc meklēšanas mēs noskaidrojām, ka vēlākajām Airflow versijām ir tā sauktais Eksperimentālais API. Vārds eksperimentāls, protams, skan biedējoši, bet ko darīt… Varbūt izdosies.

Tālāk aprakstīsim visu ceļu: no Airflow uzstādīšanas līdz POST pieprasījuma izveidošanai, kas aktivizē DAG, izmantojot Eksperimentālo API. Strādāsim ar Ubuntu 16.04.

1. Airflow uzstādīšana

Pārbaudīsim, vai mums ir instalēts Python 3 un virtualenv.

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

Ja kaut kas no tā nav, tad instalējiet.

Tagad izveidosim katalogu, kurā turpināsim strādāt ar Airflow.

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

Uzstādām Airflow:

(venv) $ pip install airflow

Versija, ar kuru mēs strādājām: 1.10.

Tagad mums ir jāizveido katalogs airflow_home, kur atradīsies DAG faili un Airflow spraudņi. Pēc kataloga izveides iestatīsim vides mainīgo AIRFLOW_HOME.

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

Nākamais solis — izpildām komandu, kas izveidos un inicializēs datu plūsmas datubāzi SQLite:

(venv) $ airflow initdb

Datu bāze tiks izveidota airflow.db pēc noklusējuma.

Pārbaudām, vai Airflow ir uzstādīts:

$ 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
  ____________       _____________
 ____    |__( )_________  __/__  /________      __
____  /| |_  /__  ___/_  /_ __  /_  __ _ | /| / /
___  ___ |  / _  /   _  __/ _  / / \/\_\/ |/ |/ /\n _/_/_  |_/_/_/  /_/_/    /_/_/    /_/_/  ____/____|__/
   v1.10.0

Ja komanda ir izpildījusies, tad Airflow ir izveidojis savu konfigurācijas failu airflow.cfg par AIRFLOW_HOME:

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

Airflow ir tīmekļa saskarne. To var palaist, izpildot komandu:

(venv) $ airflow webserver --port 8081

Tagad jūs varat piekļūt tīmekļa saskarnei pārlūkā uz porta 8081 tajā hostā, kur Airflow tika palaists, piemēram: <hostname:8081>.

2. Darbs ar Experimental API

Ar to Airflow ir konfigurēts un gatavs darbam. Tomēr mums ir jāaktivizē arī Experimental API. Mūsu pārbaudītāji ir uzrakstīti Python valodā, tāpēc visi turpmākie pieprasījumi būs uz šīs valodas pamata, izmantojot bibliotēku requests.

Patiesībā API jau darbojas vienkāršiem pieprasījumiem. Piemēram, šāds pieprasījums ļauj pārbaudīt tā darbību:

>>> import requests
>>> host = 
>>> airflow_port = 8081 #mūsu gadījumā šāds, bet noklusējumā 8080
>>> requests.get('http://{}:{} /{}'.format(host, airflow_port, 'api/experimental/test').text
'OK'

Ja saņēmāt šādu atbildi, tas nozīmē, ka viss darbojas.

Tomēr, kad mēs vēlēsimies aktivizēt DAG, mums būs jāsaskarās ar to, ka šāds pieprasījums nav iespējams veikt bez autentifikācijas.

Tāpēc būs jāveic vēl daži soļi.

Pirmkārt, konfigurācijā jāievada šis:

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

Pēc tam ir jāizveido savs lietotājs ar administratora tiesībām:

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

Pirmkārt, ir jāizveido lietotājs ar parastām tiesībām, kuram tiks atļauts izsaukt DAG trigeri.

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

Tagad viss ir gatavs.

3. POST pieprasījuma izpilde

POST pieprasījums izskatīsies šādi:

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

Pieprasījums veiksmīgi apstrādāts.

Tātad, pēc tam mēs dodam kādu laiku DAG, lai apstrādātu, un veicam pieprasījumu uz ClickHouse tabulu, mēģinot noķert kontrolpaketi.

Pārbaude pabeigta.

Avots: habr.com