Hogyan automatizálhatjuk a kódunkat Airflow DAG segítségével?

Az Airflow DAG segítségével történő kódautomatizálás egy hatékony eszközt kínál arra, hogy a kódot napi szinten futtassuk anélkül, hogy manuálisan kellene közbeavatkoznunk. Ehhez néhány lépés végrehajtására van szükség, amelyek középpontjában az Airflow eszközei és az operátorok állnak. Az alábbiakban bemutatott példa végigvezet minket ezen a folyamaton, ugyanakkor figyelembe kell venni, hogy minden egyes projekt más-más igényeket támaszt, így mindig szükséges a környezethez igazodó testreszabás.

python
Copy code
import json
import pathlib
import airflow
import requests
import requests.exceptions as request_exceptions
from datetime import date
from airflow import DAG
from airflow.operators.bash import BashOperator
from airflow.operators.python import PythonOperator
from airflow.decorators import task
from datetime import datetime, timedelta
dag_owner = 'Kendrick'

python
Copy code
def _get_pictures():
    pathlib.Path("/tmp/images").mkdir(parents=True, exist_ok=True)
    api_key = 'xxxxxxxxxxxxxxxxxxxxxxxxxxxxx'
    url = f'https://api.nasa.gov/planetary/apod?api_key={api_key}'
    response = requests.get(url).json()
    today_image = response['hdurl']

    with open(f'todays_image_{date.today()}.png', 'wb') as f:

        f.write(requests.get(today_image).content)

Miután létrehoztuk a szükséges funkciókat, meghatározzuk a DAG alapértelmezett argumentumait, amelyek biztosítják, hogy a különböző feladatok megfelelően működjenek. Az alapértelmezett paraméterek közé tartozik például a visszaállítások száma és a késleltetés, ha egy feladat nem fut le sikeresen:

python
Copy code
default_args = {

    'owner': dag_owner,
    'depends_on_past': False,
    'retries': 2,
    'retry_delay': timedelta(minutes=5)
}

A következő lépésben a DAG objektumot egy with kulcsszóval inicializáljuk, amely biztosítja, hogy a scheduler ne próbálja újra inicializálni a DAG-ot minden egyes futtatáskor. A megfelelő paraméterek beállítása után definiáljuk a feladatokat:

python
Copy code
with DAG(
    dag_id='download_ASOD_image',
    default_args=default_args,
    description='download and notify',
    start_date=airflow.utils.dates.days_ago(0),
    schedule_interval='@daily',
    catchup=False,
    tags=['None']
):
    get_pictures = PythonOperator(
        task_id="get_pictures",
        python_callable=_get_pictures
    )
    notify = BashOperator(
        task_id="notify",
        bash_command='echo "Image for today has been added!"'
    )
    get_pictures >> notify

A catchup paraméter különösen fontos, mert meghatározza, hogy a DAG újra végrehajtódjon-e a múltban elmaradt futtatások miatt. Ha például a kezdő dátumot 7 nappal korábbra állítjuk, akkor, ha a catchup értéke igaz, a rendszer az összes elmaradt napi futtatást elvégzi. A tags lehetőséget pedig arra használhatjuk, hogy a DAG-okat kategorizáljuk, így könnyebben kezelhetők a nagyobb rendszerekben.

Fontos, hogy bár a fenti kód első pillantásra bonyolultabbnak tűnhet, mint egy egyszerű Jupyter Notebook kód, valójában a legtöbb kódrészlet ismétlődő és sablon jellegű, amelyet minden Airflow DAG esetében meg kell határozni. A kód többszörös ismétlésével az Airflow képes a feladatokat automatizálni és párhuzamosítani, ezáltal gyorsabb és hatékonyabb munkafolyamatokat biztosítva.

Hogyan automatizálhatjuk kódunkat Airflow-val a DAG segítségével?

Az Airflow, egy erőteljes eszköz a munkafolyamatok automatizálására, különösen akkor, ha egy sor bonyolult feladatot kell végrehajtani egy-egy workflow-ban. A Directed Acyclic Graph (DAG) alapja az Airflow-ban az a rendszer, amelyben a feladatokat (task) felépítjük, egymás után hajtjuk végre és szükség esetén újraindítjuk őket. A DAG-ok kezelése gyakran változik, mivel a csapatok és a felelősségi körök módosulnak, ezért ajánlott időnként áttekinteni a DAG-ok és a feladatok kezelését, különösen a hibaelhárítás és az üzemeltetési felelősségek meghatározása terén.

A depends_on_past beállítás lehetőséget ad arra, hogy megakadályozza egy feladat futtatását, ha az előző feladat nem sikerült. Ez különösen fontos olyan rendszerekben, ahol az API-k nem megbízhatóak, és biztosítja, hogy a DAG akkor is futni fog, ha az előző napi futás sikertelen volt. A retries beállítás határozza meg, hogy hányszor próbálja meg újra a rendszer a feladatot, ha az nem sikerült. A példában ezt 2-re állítjuk, így ha egy feladat hibát jelez, két további próbálkozás történik, mielőtt feladja. Az Airflow alapértelmezés szerint 0-ra állítja ezt az értéket, de érdemes engedélyezni, ha az API vagy a kapcsolat nem stabil. A retry_delay beállítás 5 percre van állítva, ami biztosítja, hogy 5 perc várakozási idő legyen a feladat újraindítása előtt. Ez segít abban, hogy ne terheljük túl az API-t, ha éppen más kéréseket dolgoz fel.

Az operátorok az Airflow egyik legfontosabb elemei. Ezek encapsulálják a logikát, amely egy adott feladat elvégzéséhez szükséges, például egy Python függvény futtatása, egy Bash szkript végrehajtása vagy e-mailek küldése. Az operátorok újrahasznosíthatók és testreszabhatók, így bonyolult munkafolyamatokat építhetünk fel viszonylag egyszerűen, miközben minimalizáljuk a speciális kódírás szükségességét. Az operátorok segítségével könnyedén végezhetünk el feladatokat, legyen szó fájlkezelésről, adatbázis műveletekről vagy harmadik féltől származó rendszerekhez való kapcsolódásról.

A leggyakrabban használt operátorok közé tartoznak:

• PythonOperator: Python függvények futtatása, amelyek lehetővé teszik a bonyolult logika és adatfeldolgozás beágyazását a munkafolyamatba.

• BashOperator: Bash szkriptek futtatása, amely lehetővé teszi a shell parancsok és szkriptek alkalmazását fájlkezeléshez, rendszergazdai feladatokhoz és eszközök futtatásához.

• EmailOperator: E-mailek küldése a munkafolyamatból, értesítések, riasztások vagy jelentések küldésére.

• SimpleHttpOperator: HTTP kérések küldése külső API-khoz vagy webszolgáltatásokhoz, harmadik féltől származó rendszerek integrálása.

• SnowflakeOperator: SQL lekérdezések futtatása Snowflake adatbázisokon, adattranszformációk és elemzések elvégzése.

Az operátorok segítenek az Airflow-ban egy-egy feladat végrehajtásában, és leegyszerűsítik a feladatok összehangolását egy-egy nagyobb munkafolyamatban.

A második feladat, miután a kép sikeresen letöltődött és elmentésre került, az, hogy értesítsük a felhasználókat a folyamat befejezéséről. Ezt a BashOperator segítségével tehetjük meg, amely lehetővé teszi Bash parancsok végrehajtását. A parancs, amelyet az értesítéshez használunk: 'echo f"Image for today have been added!"', egyszerű üzenetet küld a rendszerbe, amelyet az Airflow naplóiban találunk meg.

Hogyan valósítható meg a többbérlős (multi-tenant) környezet az Apache Airflow-ban?

Az Apache Airflow egy rendkívül rugalmas és erőteljes eszköz az automatizálásra és munkafolyamatok kezelésére, de a többbérlős környezetek támogatása nem az alapértelmezett funkciói között szerepel. Azonban, ha megfelelően konfiguráljuk, akkor a több felhasználó számára izolált környezetet hozhatunk létre, amelyek képesek saját munkafolyamataik kezelésére anélkül, hogy egymás adataihoz hozzáférnének. Ezen a ponton merül fel a kérdés, hogy hogyan érhetjük el a megfelelő szintű izolációt és biztonságot.

A Kubernetes executor használata

Az egyik legegyszerűbb módja annak, hogy az Airflow-t izolált környezetekben futtassuk, a Kubernetes executor alkalmazása. Mivel minden egyes folyamat saját Pod-ban fut, a Kubernetes executor biztosítja a legmagasabb szintű izolációt. Ez azt jelenti, hogy ha a munkafolyamatok izolálása kiemelt fontossággal bír, akkor a Kubernetes executor az ideális választás. Ez a megoldás rendkívül egyszerű, és nem igényel komplex konfigurációkat. Ha azonban a biztonságos környezetekhez speciális biztonsági kontextusokra van szükség, akkor a Pod override, a node selector taints vagy a különböző képek használata segíthet a kívánt cél elérésében.

Scheduler és triggerer

Az Airflow jelenlegi architektúrája alapján a scheduler és a triggerer komponensek nem képesek többbérlős környezetek kezelésére. Mindig teljes hozzáféréssel rendelkeznek az összes DAG-hoz, és felelősek azok ütemezéséért. Ha bármilyen okból szükség van az izolációra ezen komponensek között, az egyetlen megoldás több Airflow telepítés futtatása. Azonban érdemes figyelembe venni, hogy ez a megoldás komplexitást ad hozzá a rendszerhez.

DAG-ok kezelése

Az Airflow-ban a DAG-ok egyetlen könyvtárban találhatóak, és mivel az Airflow és a Python belső működése miatt ezek a DAG-ok egy közös névtérben töltődnek be futás közben, nincs lehetőség szoros izolációra. Azonban a “push” és “pull” metodikák alkalmazásával könnyedén elérhetjük, hogy egy Airflow példányon belül különböző, elosztott DAG-ok működjenek. Bár ezek a DAG-ok nem “tudják” egymás létezését, a Python alapú importok és fájlrendszeri hozzáférés révén az egyik DAG-ban lévő információk a másik DAG számára is elérhetőek lehetnek. Ez azt jelenti, hogy ha szeretnénk izolált környezetet létrehozni a különböző felhasználók számára, akkor más megoldásokat kell keresnünk.

Web UI és többbérlős hozzáférés

Az Airflow web UI-ja támogatja a többbérlős hozzáférést szerepkörök és jogosultságok segítségével. A felhasználóknak különböző szerepköröket rendelhetünk, amelyek meghatározzák, hogy mely Airflow objektumokat érhetik el és milyen jogosultságokkal rendelkeznek. A jogosultságok párosítva vannak a főbb Airflow objektumokkal (mint például DAG, Task, Connection, Variable) és az ezekkel kapcsolatos műveletekkel (olvasás, írás, szerkesztés, törlés). A szerepkörök segítségével könnyedén finomhangolhatjuk a felhasználók hozzáférését, és biztosíthatjuk, hogy csak azok férjenek hozzá a megfelelő erőforrásokhoz.

Szerepkörök és jogosultságok kezelése

Az Airflow rendelkezik alapértelmezett szerepkörökkel, amelyek az alábbiak:

• Public

• Viewer

• User

• Op

• Admin

Ezek a szerepkörök különböző szintű hozzáférést biztosítanak az Airflow objektumokhoz. Azonban lehetőség van egyedi szerepkörök létrehozására is, amelyek az adott szervezet igényeihez igazíthatók. A szerepköröket és jogosultságokat úgy kell konfigurálni, hogy a felhasználók csak azokat a DAG-okat érhessék el, amelyekhez jogosultságuk van. Az Airflow forráskódjában található a security.permission modul, amely részletes információkat nyújt a szerepkörök és jogosultságok konfigurálásáról.

A felhasználók és szerepkörök kezelése programozott módon

A következő kódrészlet bemutatja, hogyan hozhatunk létre új felhasználókat és rendelhetjük hozzájuk a szükséges szerepköröket:

python
Copy code
def __add_users(user_name, password, email, first_name ='', last_name='', role='Public'):

    with get_application_builder() as appbuilder:
        role = appbuilder.sm.find_role(role)
        if appbuilder.sm.find_user(user_name):
            print(f"{user_name} already exists in the db")
            return
        user = appbuilder.sm.add_user(
            user_name, first_name, last_name, email, role, password
        )

Ebben a kódban a felhasználót a Public szerepkörrel hozunk létre, ami alapvetően nem biztosít hozzáférést. Az Airflow számára leginkább az adminisztrátorok kezelhetik az ilyen típusú felhasználókezelést biztonságos módon, figyelembe véve a különböző hitelesítési lehetőségeket is.

Szerepkörök hozzárendelése és jogosultságok

A következő kódrészlet bemutatja, hogyan rendelhetjük hozzá a megfelelő jogosultságokat egy felhasználóhoz és annak szerepköréhez:

python
Copy code
dag_bag = DagBag(os.path.join("dags", d))

for action in [READ, EDIT, CREATE, DELETE]:
    for d_id in dag_bag.dags.keys():
        perm: Permission | None = appbuilder.sm.create_permission(action, f"DAG:{d_id}")
        appbuilder.sm.add_permission_to_role(role, perm)

Ezen kódrészlet segítségével a megfelelő jogosultságok (olvasás, szerkesztés, létrehozás, törlés) hozzáadhatók a DAG-okhoz, és ezek a jogosultságok a megfelelő szerepkörökhöz rendelhetők. A cél, hogy a felhasználók hozzáférhessenek a számukra engedélyezett DAG-okhoz, miközben biztosítjuk, hogy nem férhetnek hozzá mások adataihoz.

Fontos szempontok és további lépések

A többbérlős konfigurációk implementálása során fontos, hogy a rendszer biztonságát és integritását ne veszélyeztessük. A jogosultságok és szerepkörök kezelése alapvetően segít abban, hogy a felhasználók kizárólag az általuk kezelt munkafolyamatokhoz férjenek hozzá. Az Airflow alapértelmezett beállításai nem biztosítanak tökéletes izolációt, ezért fontos, hogy minden felhasználónak egyedi szerepkört és jogosultságokat biztosítsunk a megfelelő izoláció érdekében.