Hogyan válasszuk a legjobb gyakorlatokat az Apache Airflow használatához?

Az Apache Airflow a modern adatfeldolgozási rendszerek egyik legfontosabb eszköze, amely a munkafolyamatok automatizálására és menedzselésére szolgál. Az Airflow lehetővé teszi a fejlesztők és adatmérnökök számára, hogy komplex adatfolyamatokat kezeljenek, amelyek több lépésből és különböző rendszerekből állnak. Az alábbiakban a legjobb gyakorlatokat és tippeket tárgyaljuk, amelyek segíthetnek az Airflow használatának hatékonyabbá tételében, és egyúttal segítenek a felhasználóknak abban, hogy jobban kihasználják az Airflow nyújtotta lehetőségeket.

Az Apache Airflow telepítése és konfigurálása előtt fontos megérteni annak alapvető összetevőit és működését. Az Airflow három alapvető komponensből áll: a DAG-ból (Directed Acyclic Graph), amely az adatfeldolgozási folyamatokat modellezi, az operátorok-ból, amelyek egy-egy feladatot vagy műveletet hajtanak végre, valamint a scheduler-ből, amely biztosítja a DAG-ok időzített futtatását. Mindezek a komponensek együttműködve biztosítják az adatok feldolgozásának folyamatos áramlását.

A legjobb gyakorlatok megértéséhez és alkalmazásához elengedhetetlen a jól megtervezett DAG-ok létrehozása. A DAG-oknak tisztán kell definiálniuk a feladatok közötti kapcsolatokat és azok futási sorrendjét. Egy rosszul megtervezett DAG nemcsak a folyamatok lelassulásához vezethet, hanem hibákhoz is, amelyek a rendszer egészére kihatnak. A DAG-ok tervezésénél figyelembe kell venni a feladatok izolációját és a megfelelő hibakezelést. A feladatok között olyan függőségeket kell kialakítani, amelyek biztosítják, hogy minden egyes lépés megfelelően végrehajtódjon, mielőtt a következő lépés elindul.

Az idempotencia, vagyis a műveletek ismételt végrehajtása ártalom nélkül, kulcsfontosságú, amikor az Airflow-t használjuk. Az idempotens feladatok biztosítják, hogy a munkafolyamatok többszöri újraindítása ne okozzon problémát vagy adatvesztést. Ha egy feladat nem idempotens, az könnyen nem kívánt mellékhatásokhoz vezethet, ha a rendszer újraindul vagy ha hibát észlel. Az idempotencia eléréséhez érdemes minden feladatot úgy megtervezni, hogy a már végrehajtott műveletek ne befolyásolják a későbbi futtatásokat.

A hibakezelés a következő legfontosabb szempont az Airflow használatakor. Az Airflow lehetőséget biztosít arra, hogy a felhasználók különféle hibaüzeneteket és állapotokat rendeljenek hozzá a feladatokhoz. Ha egy feladat nem sikerül, az Airflow automatikusan újraindíthatja azt, vagy másik feladatot futtathat, hogy biztosítsa a munkafolyamat zavartalan működését. A hibák megfelelő kezelése segít abban, hogy a rendszert hiba esetén is rugalmasan lehessen működtetni, minimális leállással.

A kód mint konfiguráció alkalmazása egy másik alapvető irányelv, amely lehetővé teszi a rugalmasságot és a skálázhatóságot. Az Airflow-ban a kódnak nemcsak a feladatok logikáját kell tartalmaznia, hanem a konfigurációs beállításokat is, például az ütemezéseket, a feladatok sorrendjét és azok paramétereit. Ezt a megközelítést segíti elő a Python alapú kódolás, amely lehetővé teszi, hogy a felhasználók könnyedén integrálják az Airflow-t más rendszerekkel és alkalmazásokkal.

Egy másik hasznos tanács az Airflow ütemezőjének és az operátoroknak a megfelelő használata. Az Airflow scheduler biztosítja, hogy a DAG-ok időben és a megfelelő sorrendben hajtódjanak végre. Fontos, hogy a felhasználók az operátorok segítségével a megfelelő típusú feladatokat válasszák ki, és azokat az ütemezésük szerint indítsák el. Az operátorok típusa meghatározza, hogy milyen típusú adatokat dolgoznak fel, például SQL lekérdezéseket, fájlokat vagy HTTP-kéréseket. A megfelelő operátor kiválasztása kritikus a teljesítmény és a skálázhatóság szempontjából.

A Kubernetes, Celery és Dask futtatókörnyezetek használata az Airflow-ban segíthet a rendszerek skálázhatóságában. A Kubernetes Executor például lehetővé teszi a feladatok futtatását konténerekben, ami különösen hasznos, ha a munkafolyamatok nagy léptékben futnak. A Celery és Dask futtatókörnyezetek szintén nagy segítséget nyújtanak a párhuzamos feldolgozásban és a feladatok elosztásában. Ezen rendszerek alkalmazása növeli a rendszer megbízhatóságát és teljesítményét, különösen akkor, ha nagy mennyiségű adatot kell feldolgozni.

Fontos, hogy a felhasználók tisztában legyenek az Airflow teljesítménybeli korlátaival is. Az Airflow megfelelő beállítása és optimalizálása érdekében elengedhetetlen a feladatok és DAG-ok hatékony ütemezése. Túl sok párhuzamosan futó feladat vagy túl nagy számú DAG egyidőben történő futtatása csökkentheti a rendszer teljesítményét. Az optimális beállítások megtalálása és az Airflow monitorozása folyamatos feladat, amely biztosítja a rendszer stabil működését.

Az Apache Airflow használata tehát összetett feladat, amely különös figyelmet igényel a tervezés, a hibakezelés, az optimalizálás és a skálázhatóság terén. A legjobb gyakorlatok alkalmazásával azonban jelentősen javíthatjuk a rendszerek megbízhatóságát és hatékonyságát. Az Airflow egy erőteljes eszköztár, amely a megfelelő tudással és gyakorlatokkal képes kiválóan támogatni az adatfolyamatok automatizálását és optimalizálását.

Milyen típusú végrehajtók állnak rendelkezésre az Apache Airflow-ban, és melyek az alkalmazásuk legjobb esetek?

A végrehajtók típusai közé tartozik a SequentialExecutor, LocalExecutor, CeleryExecutor, CeleryKubernetesExecutor, DaskExecutor, KubernetesExecutor és LocalKubernetesExecutor. Mindegyik típusnak megvannak a maga előnyei és hátrányai, amelyek különböző üzleti célok és igények függvényében hasznosak lehetnek. Minden egyes végrehajtó más-más komplexitást és skálázhatóságot kínál, így a megfelelő végrehajtó kiválasztása kulcsfontosságú a sikeres üzemeltetéshez.

A SequentialExecutor az alapértelmezett végrehajtó az Apache Airflow-ban, amely a legkevesebb erőforrást igényli, de kizárólag egyetlen feladatot képes végrehajtani egy időben. Ez ideális a helyi fejlesztési környezetekhez, vagy egyszerűbb tesztelési célokra. Az ilyen típusú végrehajtó a legjobb választás, ha nincs szükség párhuzamos feladatvégrehajtásra, és minden feladatot sorban kell végrehajtani. A SequentialExecutor leginkább kis vagy nem túl összetett adatfolyamatokhoz ajánlott.

A LocalExecutor a SequentialExecutor egy fejlettebb változata, amely lehetővé teszi a feladatok párhuzamos végrehajtását egyetlen gépen, ahol az ütemező is fut. A LocalExecutor a Python multiprocessing modulját használja, hogy több folyamatot indítson el, így nagyobb párhuzamosságot biztosít. A helyi fejlesztési környezetek számára ideális, ha nem szükséges bonyolult infrastruktúra, és kisebb, helyi környezetben kell tesztelni vagy fejleszteni. A párhuzamosság itt a gép erőforrásaitól függ, és beállítható, hogy hány feladatot hajtson végre egyszerre.

A CeleryExecutor egy másik fontos típus, amely akkor válik hasznossá, ha a feladatokat több gépen vagy munkás folyamaton keresztül kell végrehajtani. A CeleryExecutor lehetővé teszi a feladatok párhuzamos futtatását egy vagy több gépen keresztül, és különféle üzenetkezelő rendszereket igényel, mint például RabbitMQ vagy Redis. Ez a végrehajtó akkor válik szükségessé, ha nagyobb méretű adatfolyamatokat kell kezelni, és a skálázhatóság fontos szempont.

A KubernetesExecutor a Kubernetes infrastruktúrát használja a feladatok futtatásához, lehetővé téve minden feladat végrehajtását saját podban egy Kubernetes klaszterben. Ez a végrehajtó akkor jön jól, amikor a feladatokat szét kell osztani különböző gépeken vagy konténereken a nagyobb skálázhatóság és izoláció érdekében. Ez a legalkalmasabb a nagy teljesítményű környezetekhez, ahol az erőforrások dinamikusan kezelhetők és skálázhatók.

A DaskExecutor és a CeleryKubernetesExecutor különösen olyan alkalmazásokhoz hasznosak, ahol a párhuzamos számításokra van szükség, és a rendszernek képesnek kell lennie a terhelés elosztására különböző gépek között. A DaskExecutor képes kezelni a párhuzamos adatfeldolgozást elosztott környezetekben, míg a CeleryKubernetesExecutor a Kubernetes klaszterekben történő skálázást biztosít. Ezek a végrehajtók ideálisak, ha a feladatok számítási igényei nagyok, és a futtatásukhoz több erőforrás szükséges.

A végrehajtók típusainak és azok alkalmazási eseteinek megértése kulcsfontosságú ahhoz, hogy az Airflow-t hatékonyan és eredményesen lehessen alkalmazni. A megfelelő végrehajtó kiválasztása nemcsak a feladatok végrehajtásának gyorsaságát és hatékonyságát befolyásolja, hanem az egész adatfeldolgozási környezet skálázhatóságát és karbantartását is.

Fontos megérteni, hogy minden végrehajtó típusnak megvannak a maga előnyei és korlátai. A helyi környezetekben és kisebb adatfolyamatok esetén a SequentialExecutor vagy a LocalExecutor elegendő lehet, míg a nagyobb adatfolyamatok, skálázhatóság és hibatűrés esetén a CeleryExecutor, a KubernetesExecutor vagy a DaskExecutor lehet a legjobb választás. Az Airflow végrehajtóinak megértése segít abban, hogy a rendszer igényeinek leginkább megfelelő végrehajtót válasszuk, amely biztosítja a kívánt teljesítményt és hatékonyságot az adatfolyamatok kezelésében.

Hogyan kezeljük az Airflow kapcsolatait és titkos adatokat?

A kapcsolatok egyes típusainál felhasználónév és jelszó szükséges a forráshoz való csatlakozáshoz. Ha a felhasználónév és a jelszó az Airflow UI-ban kerül megadásra, az adatokat elmenti az Airflow metadabázisában. Az összes érzékeny adat titkosítva van egy Fernet kulccsal, mely rotálható. Fontos megjegyezni, hogy a jelszó automatikusan maszkolva lesz, de nem titkosítják, amikor azt az UI-ban adjuk meg. Ha a kapcsolat mentésre kerül, a felhasználó már nem láthatja a jelszót a beállítások között, mivel az Airflow automatikusan elrejti a jelszót és egyéb érzékeny adatokat, például a változókat és kulcsokat, a feladatnaplókban is.

A kapcsolatok létrehozása és szerkesztése nemcsak a felhasználói felületen történhet, hanem az Apache Airflow parancssori felületén (CLI) is. Az Airflow CLI-ben az autentikált felhasználók közvetlenül hozzáadhatják a kapcsolatokat az adatbázishoz, akár JSON, URI formátumban, vagy a paraméterek egyéni megadásával. Az Airflow CLI használata jelentős előnyt biztosít a kapcsolatok gyors szerkesztésében, új kapcsolatok létrehozásában, és az összes kapcsolat exportálásában is, ami különösen hasznos migrációk során, amikor az Open Source Airflow környezetből egy teljesen menedzselt szolgáltatásba migrálunk. Az exportált kapcsolatokat JSON, YAML vagy env formátumban is kinyerhetjük, a következő parancsok segítségével:

bash
Copy code
airflow connections export connections.json

airflow connections export /tmp/connections --file-format yaml

A kapcsolatok tesztelése fontos lépés a megfelelő működés biztosításában. Nincs rosszabb érzés, mint amikor egy hibaüzenetet kapunk egy sikertelen feladatról, és kiderül, hogy azt egy helytelen paraméter okozta. A kapcsolatok tesztelésére az egyik legnépszerűbb eszköz a Postman, amely lehetővé teszi API-kapcsolatok tesztelését. Az Airflow-ban a tesztelés alapértelmezetten ki van kapcsolva biztonsági okokból, de az Airflow konfigurációs fájl (airflow.cfg) módosításával engedélyezhetjük azt. A test_connection mezőt "Disabled"-ról "Enabled"-re kell állítani. Ha a tesztelés engedélyezve van, a felhasználó a UI vagy CLI felületen tesztelheti a kapcsolatokat. A tesztelés során az Airflow a kiválasztott hook osztály test_connection metódusát hívja meg, és az eredményeket jelenteni fogja.

Fontos azonban, hogy a biztonság mindig kiemelt figyelmet igényel. Az érzékeny adatok, például felhasználónevek, jelszavak vagy hozzáférési kulcsok védelme alapvető fontosságú a kapcsolatkezelés során. Emellett érdemes figyelembe venni, hogy a tesztelési funkció nem működik olyan kapcsolatoknál, amelyek környezeti változókban vagy külső titkos kezelőkben vannak tárolva. Ezeket a kapcsolatokat nem lehet közvetlenül tesztelni az Airflow UI vagy CLI segítségével. Alternatív megoldásként létrehozhatunk egy dummy DAG-ot, amely egy feladatban használja a kapcsolatot, így ellenőrizhetjük, hogy a kapcsolat működik.

Környezeti változók segítségével is tárolhatunk titkokat, de ez a megoldás nem biztonságos, és csak kis, ideiglenes tesztkörnyezetekben javasolt. A környezeti változók az operációs rendszer szintjén vannak beállítva, és bárminemű folyamat hozzáférhet ezekhez, amely az adott rendszerben fut. Ha ezt a megoldást választjuk, a titkokat az Airflow konfigurációs fájlban (airflow.cfg) vagy egy .env fájlban kell megadni. Azonban ez nem biztosít megfelelő védelmet, különösen nagy környezetekben, ahol több csapattag dolgozik, mivel komoly biztonsági kockázatokkal járhat.

Az Airflow metadabázisa szintén lehetőséget biztosít a titkok tárolására, és itt a titkos adatok automatikusan titkosítva vannak a Fernet kulccsal. Ez a megoldás megfelelő lehet a kezdő csapatok számára, de nem feltétlenül felel meg a biztonsági előírásoknak és nem skálázódik jól, ha a csapat növekszik.

Hogyan készíthetünk egyedi Airflow szolgáltatót és operátorokat a tea- és kávéfőzőhöz

A teás- és kávéfőzőhöz tartozó egyedi Airflow szolgáltató és operátor létrehozása során számos kulcsfontosságú lépést kell figyelembe venni. Az Airflow egy kiváló eszköz a munkafolyamatok automatizálásához, különösen akkor, ha egyedi integrációra van szükség más rendszerekkel, például egy teás- vagy kávéfőző rendszerrel. A következő részletes útmutató bemutatja, hogyan hozhatunk létre egyedi kapcsolatokat, hook-okat és operátorokat a teás és kávéfőző alkalmazások számára.

Először is, szükséges két új mezőt létrehozni a kapcsolat típusához: pot_designator és additions. Ezeket az értékeket az extra szótárban kell tárolni, amely a kapcsolat objektumhoz kapcsolódik. A hook osztályok segítségével képesek vagyunk ezeket az adatokat elérni és manipulálni az Airflow rendszerén belül. Az alábbi kódrészlet bemutatja, hogyan állíthatjuk be a mezők viselkedését a felhasználói felületen.

A következő kód segítségével a UI mezők megjelenítési módját állíthatjuk be:

python
Copy code
@staticmethod
def get_ui_field_behaviour():
    return {

        "hidden_fields": ["password", "login", "schema", "extra"],

        "placeholders": {
            "pot_designator": "1",
            "additions": "sugar",
            "host": "tea-pot",
            "port": "8083",
        },
        "relabeling": {}
    }

Ez a kód módosítja a felhasználói felületet úgy, hogy elrejti a megadott mezőket, és példákat ad a mezők kitöltéséhez. A relabeling mező ugyan üresen marad, de lehetőség van arra, hogy átnevezzük a meglévő mezőket (például a schema mezőt átnevezzük protocol-ra a felhasználói felületen, miközben az adatbázisban megőrizzük az eredeti schema attribútumot).

A kapcsolat adatainak eléréséhez és az új mezők felhasználásához a következő metódust alkalmazhatjuk:

python
Copy code
@cached_property

def get_conn(self):
    conn = self.get_connection(self.tea_pot_conn_id)
    self.url = f"{conn.host}:{conn.port}"
    self.pot_designator = conn.extra_dejson.get("pot_designator", None)
    self.additions = conn.extra_dejson.get("additions", None)
    return self

Ez a metódus a kapcsolatot egy adatbázisból szerzi meg az ID alapján, majd beállítja a megfelelő attribútumokat, mint például a pot_designator és additions.

Miután a kapcsolatot sikeresen beállítottuk, a következő lépés egy kapcsolat tesztelése. A test_connection metódus a következő módon működik:

python
Copy code
def test_connection(self):
    """Test a connection"""
    if self.is_ready():
        return (True, "Alive")
    return (False, "Not Alive")

Ez a metódus minden alkalommal fut, amikor a felhasználó az Airflow UI-ban megnyomja a Test gombot. Itt a kapcsolat állapotát ellenőrizzük, és visszaküldjük az eredményt.

A teás- és kávéfőző szolgáltató regisztrálása az Airflow-ban a következő lépés. Ehhez hozzá kell adnunk egy új mezőt a szolgáltató modulunkhoz:

python
Copy code
"connection-types": [

    {
        "connection-type": "teapot",
        "hook-class-name": "airflow_provider_tea_pot.hooks.TeaPotHook"
    }
]

Ez tájékoztatja az Airflow-t arról, hogy egy új kapcsolat típust regisztráltunk (ebben az esetben a teapot típusú kapcsolatot), és hogy az a csomagban található TeaPotHook osztályt használja.

A következő lépés a szolgáltatásaink elérésére szolgáló metódusok megírása. Itt egy egyszerűsített példát láthatunk a vízszint ellenőrzésére, a tea elkészítésére és a kávé főzésére:

python
Copy code
def make_tea(self):
    self.get_conn
    response = requests.get(f"http://{self.url}/ready")
    return self.__check_and_return(response)
def brew_coffee(self):
    self.get_conn
    response = requests.get(f"http://{self.url}/ready")
    return self.__check_and_return(response)
def get_water_level(self):
    self.get_conn
    response = requests.get(f"http://{self.url}/ready")
    return self.__check_and_return(response)

Ebben a példában egy egyszerű metódust alkalmazunk minden egyes kimenethez (tea, kávé, vízszint), amely ellenőrzi a 200-as HTTP választ, és hibát dob, ha nem sikerült válaszolni.

Miután elkészítettük a hookokat, rátérhetünk a saját operátoraink megírására. Az operátorok alapvető szerkezete az alábbiak szerint néz ki:

python
Copy code
class MakeTeaOperator(BaseOperator):

    def __init__(self, tea_pot_conn_id, additions=None, pot_designator=None, **kwargs) -> None:
        super().__init__(**kwargs)
        self.tea_pot_conn_id = tea_pot_conn_id
        self.pot_designator = pot_designator
        self.additions = additions
    def execute(self, context) -> typing.Any:
        self.hook = TeaPotHook(tea_pot_conn_id=self.tea_pot_conn_id)
        if self.pot_designator:
            self.hook.pot_designator = self.pot_designator
        if self.additions:
            self.hook.additions = self.additions
        return self.hook.make_tea()

Itt az init metódusban tároljuk a konfigurációs információkat, míg az execute metódus a hookot használja a tea elkészítéséhez.

A trigger kialakítása az alábbiak szerint történik:

python
Copy code
class WaterLevelTrigger(BaseTrigger):

    def __init__(self, tea_pot_conn_id, minimum_level) -> None:

        self.tea_pot_conn_id = tea_pot_conn_id
        self.minimum_level = minimum_level
    def serialize(self) -> typing.Tuple[str, typing.Dict[str, typing.Any]]:
        return "airflow_provider_tea_pot.triggers.WaterLevelTrigger", {
            "minimum_level": self.minimum_level,
            "tea_pot_conn_id": self.tea_pot_conn_id
        }
    async def run(self):
        hook = TeaPotHook(tea_pot_conn_id=self.tea_pot_conn_id)
        async_get_water_level = sync_to_async(hook.get_water_level)
        while True:
            rv = await async_get_water_level()
            if json.loads(rv).get('level') > self.minimum_level:
                yield TriggerEvent(rv)

Ez a trigger folyamatosan ellenőrzi a vízszintet, és ha az meghaladja a minimum szintet, aktiválja a megfelelő eseményt.

A triggerek és deferálható operátorok tervezése során fontos, hogy a kód ne blokkolja a végrehajtási szálakat, és gondoskodjunk a megfelelő hibakezelésről.