🥇How to trigger a DAG in Airflow using the Experimental API

Når vi forbereder vores uddannelsesprogrammer, støder vi fra tid til anden på udfordringer i forhold til værktøjerne. Og på det tidspunkt, hvor vi møder dem, er der ikke altid tilstrækkelig dokumentation og artikler, der kan hjælpe os med at håndtere disse problemer.

Det var for eksempel tilfældet i 2015, hvor vi i programmet "Specialist i store data" brugte en Hadoop-klynge med Spark til 35 samtidige brugere. Hvordan man forbereder det til denne brugssag med YARN, var uklart. Til sidst, efter at have arbejdet os igennem problemet selv, gjorde vi et indlæg på Habr og vi præsenterede endnu engang på Moscow Spark Meetup.

Background

Denne gang vil vi tale om et andet program – Data Engineer. Her bygger vores deltagere to typer arkitektur: lambda og kappa. I lambda-arkitekturen anvendes Airflow til batchbehandling af logs fra HDFS til ClickHouse.

Alt er egentlig godt. Lad dem bygge deres pipelines. Dog er der et men: alle vores programmer er teknologisk avancerede i forhold til selve læringsprocessen. Til kontrol af labben bruger vi automatiske checkere: deltageren skal logge ind på sin konto, klikke på knappen "Tjek", og efter lidt tid ser han en slags udvidet feedback på det, han har gjort. Og netop i det øjeblik begynder vi at nærme os vores problem.

Kontrollen af denne lab er sat op således: vi sender et kontroldatapakke til deltagerens Kafka, derefter overfører Gobblin dette datapakke til HDFS, og derefter tager Airflow dette datapakke og placerer det i ClickHouse. Pointen er, at Airflow ikke må gøre dette i realtid, det gør det efter en tidsplan: hver 15. minut tager det en bunke filer og sender dem.

Det viser sig, at vi på en eller anden måde skal trigger deres DAG selv efter vores behov under arbejdet med checkeren her og nu. Ved at google fandt vi ud af, at der for nyere versioner af Airflow findes noget, der kaldes Experimental API. Ordet experimental, lyder selvfølgelig skræmmende, men hvad skal man gøre… Måske lykkes det.

Vi vil nu beskrive hele processen: fra installationen af Airflow til dannelsen af POST-anmodningen, der trigger DAG, ved hjælp af Experimental API. Vi vil arbejde med Ubuntu 16.04.

1. Installation af Airflow

Lad os tjekke, om vi har Python 3 og virtualenv installeret.

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

Hvis nogen af disse mangler, så installer dem.

Lad os nu oprette en mappe, hvor vi vil arbejde videre med Airflow.

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

Lad os installere Airflow:

(venv) $ pip install airflow

Versionen, vi arbejdede med: 1.10.

Nu skal vi oprette en mappe airflow_home, hvor DAG-filerne og Airflows plugins vil blive placeret. Efter oprettelsen af mappen sætter vi miljøvariablen AIRFLOW_HOME.

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

Næste skridt er at udføre kommandoen, som vil oprette og initialisere databasen for dataflowet i SQLite:

(venv) $ airflow initdb

Databasen vil blive oprettet i airflow.db Opgave

Lad os kontrollere, om Airflow er installeret:

$ 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 lykkedes, har Airflow oprettet sin konfigurationsfil airflow.cfg i AIRFLOW_HOME:

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

Airflow har et webinterface. Det kan startes ved at køre følgende kommando:

(venv) $ airflow webserver --port 8081

Nu kan du tilgå webinterfacet i din browser på port 8081 på den vært, hvor Airflow blev startet, for eksempel: <hostname:8081>.

2. Arbejde med Experimental API

Nu er Airflow konfigureret og klar til brug. Men vi skal også starte den Experimental API. Vores checkere er skrevet i Python, så alle efterfølgende anmodninger vil være på det sprog ved hjælp af biblioteket requests.

API'en fungerer allerede for enkle anmodninger. For eksempel tillader denne anmodning at teste det:

>>> import requests
>>> host = 
>>> airflow_port = 8081 #i vores tilfælde, men som standard 8080
>>> requests.get('http://{}:{} /{}'.format(host, airflow_port, 'api/experimental/test').text
'OK'

Hvis du modtog denne meddelelse som svar, betyder det, at alt fungerer.

Men når vi ønsker at trigge DAG, vil vi støde på, at denne type anmodning ikke kan foretages uden autentifikation.

For dette skal vi udføre et par yderligere skridt.

For det første skal vi tilføje dette i konfigurationen:

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

Derefter skal vi oprette vores egen bruger med administratorrettigheder:

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

Hereafter, you need to create a user with regular permissions who will be allowed to trigger the 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()

Now everything is ready.

3. Starting a POST request

The POST request will look like this:

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

The request was processed successfully.

Accordingly, then we give some time for the DAG to process and make a request to the ClickHouse table, trying to catch the control data packet.

The check is complete.

Kilde: habr.com