Oktatási programjaink előkészítése során időnként nehézségekbe ütközünk bizonyos eszközök használatával kapcsolatban. Abban a pillanatban, amikor ezekkel a problémákkal szembesülünk, gyakran nem áll rendelkezésre elegendő dokumentáció vagy cikk, amely segítene megoldani a helyzetet.
Például 2015-ben a "Nagy adatok szakértője" program keretében Hadoop klasztert használtunk Spark-kal 35 egyidejű felhasználóval. Nem volt világos, hogyan készíthetjük elő ezt az esetet YARN használatával. Végül, miután magunk fedeztük fel a megoldásokat, és előadásokat tartottunk a .
Háttér
Ezúttal egy másik programról lesz szó – . Ezen a programon résztvevőink kétféle architektúrát építenek: lambda és kappa. A lambda architektúrában, a batch feldolgozás keretében az Airflow-t használjuk a logok HDFS-ből ClickHouse-ba történő áthelyezésére.
Alapvetően minden rendben van. Hadd építsenek pipelinet. Viszont van egy de: minden programunk technológiai szempontból a tanulási folyamat során nagyon kifinomult. A laborok ellenőrzésére automatikus ellenőrzőket használunk: a résztvevőnek be kell lépnie a fiókjába, meg kell nyomnia a "Ellenőriz" gombot, és egy idő után visszajelzést kap arról, amit tett. És pontosan ebben a pillanatban kezdjük el közelíteni a problémánkhoz.
A labor ellenőrzése a következőképpen működik: ellenőrző adatcsomagot küldünk a résztvevő Kafka-jába, majd a Gobblin áthelyezi ezt az adatcsomagot HDFS-re, ezt követően az Airflow átemeli az adatcsomagot és elhelyezi a ClickHouse-ban. A dolog lényege, hogy az Airflow-nak nem szabad valós időben végeznie ezt a folyamatot, hanem ütemezés szerint: 15 percenként egy adag fájlt vesz és elhelyez.
Tehát valahogyan szükségünk van arra, hogy mi magunk aktiváljuk a DAG-ot a vizsgáló működése közben. Utánanézve kiderült, hogy a későbbi Airflow verziókhoz létezik úgynevezett . A "kísérleti" szó természetesen ijesztően hangzik, de mit lehet tenni… Lehet, hogy működik. Ezután leírjuk az összes lépést: az Airflow telepítésétől kezdve a POST-kérdés felállításáig, amely aktiválja a DAG-ot az Experimental API használatával. Az Ubuntu 16.04-gyel fogunk dolgozni.1. Az Airflow telepítése
Ellenőrizzük, hogy a Python 3 és a virtualenv telepítve van-e.
$ python3 --version Python 3.6.6 $ virtualenv --version 15.2.0
Ha valamelyik hiányzik, telepítse.
Most hozzunk létre egy könyvtárat, amelyben az Airflow-val dolgozunk.Ha valamelyik elem hiányzik, kérjük, telepítse azt.
Most létrehozunk egy könyvtárat, amellyel később dolgozunk az Airflow-val.
$ mkdir
$ cd /path/to/your/new/directory
$ virtualenv -p which python3 venv
$ source venv/bin/activate
(venv) $Telepítsük az Airflow-t:
(venv) $ pip install airflowA verzió, amelyen dolgoztunk: 1.10.
Most létre kell hoznunk egy könyvtárat airflow_home, ahol az Airflow DAG-fájlok és bővítmények találhatók. A könyvtár létrehozása után állítsuk be a környezeti változót AIRFLOW_HOME.
(venv) $ cd /path/to/my/airflow/workspace
(venv) $ mkdir airflow_home
(venv) $ export AIRFLOW_HOME=A következő lépés — futtassuk a parancsot, amely létrehozza és inicializálja az adatbázist SQLite-ban:
(venv) $ airflow initdbAz adatbázis létrejön airflow.db alapértelmezés szerint.
Ellenőrizzük, hogy az Airflow telepítve van-e:
$ airflow version
[2018-11-26 19:38:19,607] {__init__.py:57} INFO - Használatban lévő executor: SequentialExecutor
[2018-11-26 19:38:19,745] {driver.py:123} INFO - Nyelvtani táblák generálása a /usr/lib/python3.6/lib2to3/Grammar.txt-ből
[2018-11-26 19:38:19,771] {driver.py:123} INFO - Nyelvtani táblák generálása a /usr/lib/python3.6/lib2to3/PatternGrammar.txt-ből
____________ _____________
____ |__( )_________ __/__ /________ __
____ /| |_ /__ ___/_ /_ __ /_ __ _ | /| / /
___ ___ | / _ / _ __/ _ / /_ /_ / /| / |/
_/_/ |_/_/ /_/ /_/ /_/ ____/____/|__/
v1.10.0Ha a parancs sikeresen lefutott, az Airflow létrehozta a konfigurációs fájlját airflow.cfg be. AIRFLOW_HOME:
$ tree
.
├── airflow.cfg
└── unittests.cfgAz Airflow-nak van webes felülete. Ezt a következő paranccsal indíthatjuk el:
(venv) $ airflow webserver --port 8081Most már elérheti a webes felületet a böngészőben, a 8081-es porton azon a gépen, ahol az Airflow indítva van, például: <hostname:8081>.
2. Munkavégzés az Experimental API-val
Ezen az Airflow beállítása elkészült és készen áll a munkára. Mindazonáltal el kell indítanunk az Experimental API-t is. Ellenőrzőink Python nyelven íródtak, ezért a továbbiakban minden kérdés Pythonban történik a library-k segítségével. Általános formában a web-kérdések szerkezete három részből áll:.
Valójában az API már működik egyszerű kérdésekhez. Például, ez a kérés lehetővé teszi a működés tesztelését:
>>> import requests
>>> host =
>>> airflow_port = 8081 # az esetünkben ez, alapértelmezetten 8080
>>> requests.get('http://{}:{}//{}'.format(host, airflow_port, 'api/experimental/test').text
'OK'Ha ilyen üzenetet kaptunk válaszként, az azt jelenti, hogy minden jól működik.
Ha azonban DAG-ot szeretnénk rátérni, azt tapasztalhatjuk, hogy ezt a kérést nem lehet végrehajtani autentikáció nélkül.
Ehhez még számos lépést kell követnünk.
Először is, a konfigurációhoz hozzá kell adnunk ezt:
[api]
auth_backend = airflow.contrib.auth.backends.password_authEzután létre kell hoznunk egy saját felhasználót adminisztrátori jogosultságokkal:
>>> 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()Ezután létre kell hozni egy olyan felhasználót, aki normál jogosultságokkal rendelkezik, és akinek engedélyezett a DAG triggerelése.
>>> 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()Most már minden kész.
3. POST kérelem indítása
A POST kérés így fog kinézni:
>>> 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'A kérés sikeresen feldolgozva.
Ennélfogva adjunk valamennyi időt a DAG-nak a feldolgozásra, és végezzünk el egy kérést a ClickHouse táblában, próbálva elkapni egy kontroll adatcsomagot.
Ellenőrzés befejezve.
Forrás: habr.com
