🥇Hvordan trigge en DAG i Airflow ved hjelp av Experimental API

Når vi forbereder våre utdanningsprogrammer, støter vi periodisk på utfordringer med enkelte verktøy. Og i øyeblikkene vi møter dem, er det ikke alltid tilstrekkelig dokumentasjon og artikler som kan hjelpe oss med å løse dette problemet.

Det var for eksempel i 2015, da vi i programmet "Spesialist på store data" brukte et Hadoop-kluster med Spark for 35 samtidige brukere. Hvordan man skulle forberede det for en slik brukstilfelle ved hjelp av YARN, var uklart. Til slutt, etter å ha jobbet oss gjennom det på egenhånd, fikk vi gjort et innlegg på Habré og vi holdt også et innlegg på Moscow Spark Meetup.

Bakgrunn

Denne gangen skal vi snakke om et annet program – Data Engineer. Deltakerne våre bygger to typer arkitektur: lambda og kappa. Og i lambda-arkitekturen brukes Airflow til å overføre logger fra HDFS til ClickHouse i batchbehandling.

Alt er egentlig bra. La dem bygge sine pipelines. Men det er en hake: alle programmene våre er teknologisk krevende med tanke på selve læringsprosessen. For å sjekke labene bruker vi automatiserte sjekker: deltakerne må logge inn på kontoen sin, klikke på "Sjekk", og etter en stund ser de en omfattende tilbakemelding på det de har gjort. Og nettopp i dette øyeblikket begynner vi å nærme oss vårt problem.

Sjekken av denne laben er satt opp slik: vi sender en kontrollpakke med data til deltakerens Kafka, deretter flytter Gobblin denne datapakken til HDFS, videre tar Airflow pakka og plasserer den i ClickHouse. Tingen er at Airflow ikke skal gjøre dette i sanntid; han gjør det etter en tidsplan: hver 15. minutt tar han en pakke filer og laster dem opp.

Så vi må på en eller annen måte trigge deres DAG selv etter våre krav, her og nå under sjekkerens arbeid. Etter å ha googlet det, fant vi ut at for nyere versjoner av Airflow finnes det et såkalt Experimental API. Ordet eksperimentell, høres selvfølgelig skremmende ut, men hva skal man gjøre… Kanskje det vil fungere.

Videre beskriver vi hele veien: fra installasjon av Airflow til forming av POST-forespørselen som trigget DAG ved å bruke Experimental API. Vi kommer til å jobbe med Ubuntu 16.04.

1. Installering av Airflow

La oss sjekke at vi har Python 3 og virtualenv installert.

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

Hvis noe av dette mangler, vennligst installer det.

Nå lager vi en katalog der vi skal jobbe videre med Airflow.

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

La oss installere Airflow:

(venv) $ pip install airflow

Versjonen vi jobbet med: 1.10.

Nå må vi opprette en katalog airflow_home, hvor DAG-filer og Airflow-plugins skal ligge. Etter å ha opprettet katalogen, setter vi miljøvariabelen AIRFLOW_HOME.

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

Neste steg er å kjøre kommandoen som vil opprette og initialisere databasen for datastrømmen i SQLite:

(venv) $ airflow initdb

Databasen vil bli opprettet i airflow.db som standard.

La oss sjekke om Airflow er installert:

$ 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

Hvis kommandoen fungerte, har Airflow opprettet sin konfigurasjonsfil airflow.cfg i AIRFLOW_HOME:

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

Airflow har et webgrensesnitt. Dette kan startes ved å kjøre kommandoen:

(venv) $ airflow webserver --port 8081

Nå kan du få tilgang til webgrensesnittet i en nettleser på port 8081 på vertsmaskinen der Airflow ble startet, for eksempel: <hostname:8081>.

2. Arbeide med Experimental API

Nå er Airflow konfigurert og klart til bruk. Likevel, vi må også starte det eksperimentelle API-et. Våre sjekker er skrevet i Python, så videre vil alle forespørslene være på det med bruk av biblioteket requests.

Faktisk er API-et allerede aktivt for enkle forespørslene. For eksempel, denne forespørselen kan teste det:

>>> import requests
>>> host = 
>>> airflow_port = 8081 #i vårt tilfelle er det slik, men som standard er det 8080
>>> requests.get('http://{}:{} /{}'.format(host, airflow_port, 'api/experimental/test').text
'OK'

Hvis vi fikk en slik melding som svar, betyr det at alt fungerer.

Imidlertid, når vi ønsker å trigge DAG, vil vi oppleve at denne typen forespørsel ikke kan utføres uten autentisering.

For dette må vi gjøre en rekke ytterligere trinn.

For det første, må vi legge til dette i konfigurasjonen:

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

Deretter må vi opprette en bruker med administratorrettigheter:

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

Deretter må vi opprette en bruker med vanlige rettigheter som vil få lov til å trigge DAG-en.

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

Nå er alt klart.

3. Kjøre POST-forespørselen

POST-forespørselen vil se slik ut:

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

Forespørselen ble behandlet vellykket.

Deretter gir vi DAG-en litt tid til å behandle og gjør et forespørsel i ClickHouse-tabellen, i et forsøk på å fange kontrollpakken med data.

Kontrolleringen er fullført.

Kilde: habr.com