Bei der Vorbereitung unserer Bildungsprogramme stoßen wir immer wieder auf Schwierigkeiten im Umgang mit bestimmten Werkzeugen. Und zu dem Zeitpunkt, an dem wir mit ihnen konfrontiert werden, gibt es nicht immer genügend Dokumentation und Artikel, die bei der Bewältigung dieses Problems hilfreich wären.
So war es beispielsweise im Jahr 2015, als wir im Programm "Spezialist für Big Data" einen Hadoop-Cluster mit Spark für 35 gleichzeitige Nutzer verwendeten. Wie man dies für einen solchen Anwendungsfall mit YARN vorbereitet, war unklar. Am Ende, nachdem wir uns selbst damit auseinandergesetzt hatten, haben wir und außerdem auf der .
Vorgeschichte
In dieser Runde wird es um ein anderes Programm gehen – . Dort bauen unsere Teilnehmer zwei Arten von Architekturen: Lambda und Kappa. In der Lambda-Architektur wird im Rahmen der Batch-Verarbeitung Airflow verwendet, um Logs von HDFS nach ClickHouse zu verschieben.
Im Großen und Ganzen läuft es gut. Lassen Sie sie ihre Pipelines bauen. Es gibt jedoch einen Haken: Alle unsere Programme sind technologisch anspruchsvoll in Bezug auf den Lernprozess selbst. Zur Überprüfung der Labs verwenden wir automatische Checker: Der Teilnehmer muss sich in seinen persönlichen Bereich einloggen, auf die Schaltfläche "Überprüfen" klicken, und nach einer gewissen Zeit sieht er ein erweitertes Feedback zu dem, was er gemacht hat. Und genau in diesem Moment nähern wir uns unserem Problem.
Die Überprüfung dieses Labs ist so gestaltet: Wir senden ein Kontrollpaket von Daten an Kafka des Teilnehmers, dann verschiebt Gobblin dieses Datenpaket auf HDFS, anschließend nimmt Airflow dieses Datenpaket und legt es in ClickHouse ab. Der Clou ist, dass Airflow dies nicht in Echtzeit machen soll, sondern nach einem Zeitplan: alle 15 Minuten nimmt es ein Paket von Dateien und lädt es hoch.
Das bedeutet, dass wir auf irgendeine Weise ihren DAG eigenständig nach unserem Bedarf während der Arbeit des Checkers hier und jetzt triggern müssen. Nach etwas Googeln fanden wir heraus, dass es für die späten Versionen von Airflow so etwas wie eine . Das Wort experimentell, klingt natürlich beängstigend, aber was soll man tun… Vielleicht funktioniert es ja.
Wir werden den gesamten Weg beschreiben: von der Installation von Airflow bis zur Erstellung einer POST-Anfrage, die den DAG mit Hilfe der Experimental API auslöst. Wir arbeiten mit Ubuntu 16.04.
1. Installation von Airflow
Überprüfen wir, ob wir Python 3 und virtualenv installiert haben.
$ python3 --version
Python 3.6.6
$ virtualenv --version
15.2.0Wenn etwas davon fehlt, installieren Sie es bitte.
Jetzt erstellen wir ein Verzeichnis, in dem wir weiter mit Airflow arbeiten werden.
$ mkdir
$ cd /pfad/zum/deinem/neuen/verzeichnis
$ virtualenv -p which python3 venv
$ source venv/bin/activate
(venv) $Installieren wir Airflow:
(venv) $ pip install airflowDie Version, mit der wir gearbeitet haben: 1.10.
Jetzt müssen wir ein Verzeichnis erstellen airflow_home, in dem sich die DAG-Dateien und Airflow-Plugins befinden werden. Nach der Erstellung des Verzeichnisses setzen wir die Umgebungsvariable AIRFLOW_HOME.
(venv) $ cd /pfad/zum/meinem/airflow/workspace
(venv) $ mkdir airflow_home
(venv) $ export AIRFLOW_HOME=Der nächste Schritt ist die Ausführung des Befehls, der die Datenbank für die Datenströme in SQLite erstellt und initialisiert:
(venv) $ airflow initdbDie Datenbank wird erstellt in airflow.db von Haus aus.
Überprüfen wir, ob Airflow installiert wurde:
$ airflow version
[2018-11-26 19:38:19,607] {__init__.py:57} INFO - Verwendung des Executors SequentialExecutor
[2018-11-26 19:38:19,745] {driver.py:123} INFO - Erzeuge Grammatik-Tabellen aus /usr/lib/python3.6/lib2to3/Grammar.txt
[2018-11-26 19:38:19,771] {driver.py:123} INFO - Erzeuge Grammatik-Tabellen aus /usr/lib/python3.6/lib2to3/PatternGrammar.txt
____________ _____________
____ |__( )_________ __/__ /________ __
____ /| |_ /__ ___/_ /_ __ /_ __ _ | /| / /
___ ___ | / _ / _ __/ _ / / _/_/ \/ |/ /
_/_/_/ |_/_/_/ /_/ /_/ /_/ ____/____/|__/
v1.10.0Wenn der Befehl erfolgreich war, hat Airflow seine Konfigurationsdatei erstellt airflow.cfg in AIRFLOW_HOME:
$ tree
.
├── airflow.cfg
└── unittests.cfgAirflow hat ein Web-Interface. Es kann mit folgendem Befehl gestartet werden:
(venv) $ airflow webserver --port 8081Jetzt können Sie über das Web-Interface im Browser auf Port 8081 auf dem Host zugreifen, auf dem Airflow gestartet wurde, zum Beispiel: <hostname:8081>.
2. Arbeiten mit der Experimental-API
Damit ist Airflow konfiguriert und bereit zur Nutzung. Allerdings müssen wir auch die Experimental-API starten. Unsere Checker sind in Python geschrieben, daher werden alle kommenden Anfragen in dieser Sprache mit der Bibliothek erfolgen. requests.
Tatsächlich funktioniert die API bereits für einfache Anfragen. Zum Beispiel ermöglicht folgende Anfrage, ihre Funktionsweise zu testen:
>>> import requests
>>> host =
>>> airflow_port = 8081 #in unserem Fall so, vorab 8080
>>> requests.get('http://{}:{} /{}'.format(host, airflow_port, 'api/experimental/test')).text
'OK'Wenn Sie eine solche Rückmeldung erhalten haben, bedeutet das, dass alles funktioniert.
Wenn wir jedoch versuchen, einen DAG zu triggern, werden wir feststellen, dass diese Art von Anfrage nicht ohne Authentifizierung durchgeführt werden kann.
Dafür müssen noch einige weitere Schritte unternommen werden.
Zunächst müssen wir dies in die Konfiguration hinzufügen:
[api]
auth_backend = airflow.contrib.auth.backends.password_authDann müssen wir einen eigenen Benutzer mit Administratorrechten erstellen:
>>> 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()Dann müssen wir einen Benutzer mit normalen Rechten erstellen, dem es erlaubt ist, einen DAG-Trigger auszulösen.
>>> 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()Jetzt ist alles bereit.
3. Start des POST-Requests
Der POST-Request wird folgendermaßen aussehen:
>>> 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'Die Anfrage wurde erfolgreich bearbeitet.
Entsprechend geben wir dem DAG etwas Zeit zur Verarbeitung und führen eine Anfrage an die ClickHouse-Tabelle durch, um das Kontrollpaket an Daten zu erfassen.
Überprüfung abgeschlossen.
Quelle: habr.com
