Hogyan bővíthetjük az Airflow funkcionalitását egyedi UI pluginekkel?

Az Apache Airflow egy robusztus és kiterjeszthető munkafolyamat-kezelő rendszer, amely lehetővé teszi a felhasználók számára, hogy hatékonyan automatizálják és nyomon kövessék az adatfeldolgozási munkafolyamataikat. Az Airflow alapértelmezett felhasználói felülete számos alapvető funkcióval rendelkezik, de gyakran előfordul, hogy egyedi, testreszabott megoldásokra van szükség a munkafolyamatok teljesítményének és egészségének figyelemmel kíséréséhez. Ebben a fejezetben azt fogjuk áttekinteni, hogyan hozhatunk létre egyedi Airflow UI plugineket, amelyek lehetővé teszik az egyedi mérőszámok és dashboardok megjelenítését a munkafolyamatok monitorozásához.

A saját Airflow UI plugin létrehozása különösen hasznos, ha valamilyen domain-specifikus mérőszámot, vizualizációt vagy figyelmeztető eszközt szeretnénk integrálni a rendszerbe. Az Airflow UI pluginek regisztrálása egy Python osztályban történik, amely az airflow.plugins_manager.AirflowPlugin interfészt örökli. Ezen kívül számos más testreszabási lehetőség is elérhető, például Flask alkalmazások, egyedi hook-ok, időzítők és egyéb funkciók, melyeket a fejlesztők az Airflow dokumentációjában részletesen megismerhetnek.

Airflow UI pluginek megértése

A következő kódpélda bemutatja az AirflowPlugin osztály alapjait:

python
Copy code
class AirflowPlugin:
    name = None  # A plugin neve
    hooks = []  # A plugin által használt hook-ok listája
    macros = []  # A makrók listája
    flask_blueprints = []  # Flask blueprint-ek listája
    appbuilder_views = []  # AppBuilder nézetek listája
    appbuilder_menu_items = []  # Menüpontok a plugin számára
    global_operator_extra_links = []  # Operátor szintű extra linkek
    operator_extra_links = []  # Egyedi operátor extra linkek
    timetables = []  # Időzítők listája
    listeners = []  # Eseményekhez való figyelők

Egyedi méretű dashboard plugin létrehozása

Ebben a részben egy egyszerű méretű mérőszám dashboardot készítünk, amely elérhető lesz az Airflow helyi környezetéből. A vizualizációkhoz a Chart.js könyvtárat fogjuk használni, amely egy könnyen használható JavaScript alapú diagramkészítő könyvtár. A könyvtár számos alapértelmezett diagramtípust kínál, amelyeket könnyen integrálhatunk saját alkalmazásunkba.

A következő lépésekben részletesen bemutatjuk a dashboard plugin létrehozását:

1. Projekt struktúra

plaintext
Copy code
plugins
└── metrics_plugin
    ├── __init__.py
    ├── templates
    │   └── dashboard.html
    └── views
        ├── __init__.py
        └── dashboard.py

2. Dashboard.html sablon

A sablonok (templates) az Airflow webalkalmazás részét képezik, amely HTML fájlokat tartalmaz, és az adatokat dinamikusan jeleníti meg. A dashboard.html sablon tartalmazza a mérőszámok megjelenítésére szolgáló Chart.js-diagramokat és egyéb vizualizációkat.

3. Dashboard.py nézetek

Használati esettanulmányok

A testreszabott Airflow UI plugin-ek hasznosak lehetnek számos különböző üzleti igény kielégítésére. Például:

• Metastore böngésző: Az Airflow UI plugin segítségével a felhasználók könnyedén megtekinthetik a Hive Metastore tábláinak részleteit, vagy logokat elemezhetnek.

• Monitorozási integráció: Az Airflow rendszerébe beépített monitorozó eszközként használható, például a Datadog integrálásával, így a felhasználók egy helyen kezelhetik a DAG-ok monitorozását.

• SLA státuszok vizualizálása: Egyedi plugin létrehozásával a DAG-ok SLA státuszait egyetlen dashboardon, vizuálisan jeleníthetjük meg.

További szempontok

Bár a fent bemutatott példák és megoldások hasznosak lehetnek, érdemes figyelembe venni, hogy a pluginek fejlesztése és karbantartása időigényes feladat lehet. Fontos, hogy a fejlesztők mindig frissítsék a plugineket az újabb Airflow verziók megjelenésekor, mivel az új verziók új funkciókat és javításokat hozhatnak, amelyek befolyásolhatják a pluginok működését. Az Airflow közösségi fórumain és dokumentációs oldalain érdemes követni a legújabb fejlesztéseket, hogy mindig naprakészen tudd, miként bővítheted tovább a rendszert.

Hogyan készíthetünk saját metrikák kezelőpanelt Flask és Airflow segítségével?

A következő lépésekben bemutatjuk, hogyan építhetünk egy egyszerű metrikák kezelőpanelt Flask alapú alkalmazással, amelyet az Airflow rendszeren belül használhatunk a DAG futások statisztikai adatainak megjelenítésére. A példa egy plugin modult használ, amely a backend kódot és a frontend HTML sablonokat integrálja, hogy a felhasználók valós időben hozzáférhessenek a fontos statisztikai mutatókhoz.

Backend nézet implementálása

A backend kód implementálása a views/dashboard.py fájlban történik, ahol az Airflow adatbázisból lekérdezhetjük a szükséges statisztikai adatokat. Ehhez először beállítjuk azokat az importokat, amelyek szükségesek a dashboard nézethez:

python
Copy code
from __future__ import annotations
from typing import TYPE_CHECKING
from airflow.auth.managers.models.resource_details import AccessView
from airflow.utils.session import NEW_SESSION, provide_session
from airflow.www.auth import has_access_view
from flask_appbuilder import BaseView, expose
from sqlalchemy import text
if TYPE_CHECKING:
    from sqlalchemy.orm import Session

Ezután definiáljuk a MetricsDashboardView osztályt, amely meghatározza a weboldal útvonalát:

python
Copy code
class MetricsDashboardView(BaseView):
    """A Flask-AppBuilder View for a metrics dashboard"""
    default_view = "index"
    route_base = "/metrics_dashboard"

python
Copy code
@provide_session
@expose("/")
@has_access_view(AccessView.PLUGINS)
def index(self, session: Session = NEW_SESSION):
    """Create dashboard view"""
    def interval(n: int):
        return f"now() - interval '{n} days'"
    dag_run_query = text(f"""
        SELECT
            dr.dag_id,
            SUM(CASE WHEN dr.state = 'success' AND dr.start_date > {interval(1)} THEN 1 ELSE 0 END) AS "1_day_success",
            SUM(CASE WHEN dr.state = 'failed' AND dr.start_date > {interval(1)} THEN 1 ELSE 0 END) AS "1_day_failed",
            SUM(CASE WHEN dr.state = 'success' AND dr.start_date > {interval(7)} THEN 1 ELSE 0 END) AS "7_days_success",
            SUM(CASE WHEN dr.state = 'failed' AND dr.start_date > {interval(7)} THEN 1 ELSE 0 END) AS "7_days_failed",
            SUM(CASE WHEN dr.state = 'success' AND dr.start_date > {interval(30)} THEN 1 ELSE 0 END) AS "30_days_success",
            SUM(CASE WHEN dr.state = 'failed' AND dr.start_date > {interval(30)} THEN 1 ELSE 0 END) AS "30_days_failed"
        FROM dag_run AS dr
        JOIN dag AS d ON dr.dag_id = d.dag_id
        WHERE d.is_paused != true
        GROUP BY dr.dag_id
    """)
    dag_run_stats = [dict(result) for result in session.execute(dag_run_query)]
    return self.render_template(
        "dashboard.html",
        title="Metrics Dashboard",
        dag_run_stats=dag_run_stats,
    )

Metrikai dashboard HTML sablon

Most, hogy a backend kód kész, létre kell hoznunk a HTML sablont a metrikák dashboardjának megjelenítéséhez a templates/dashboard.html fájlban. Az Airflow alapértelmezett sablonjaiban található Bootstrap CSS-t fogjuk használni, hogy a kezelőpanel reszponzív és felhasználóbarát legyen.

html
Copy code
{% extends base_template %}

{% block title %}
    {{ title }}
{% endblock %}
{% block head_meta %}
    {{ super() }}
{% endblock %}
{% block content %}
    {{ title }}
    <div class="container">
        <canvas id="successChart"></canvas>

        <canvas id="failedChart"></canvas>

    </div>
{% endblock %}
{% block tail %}
    {{ super() }}
{% endblock %}

A fenti HTML sablonban a {{ title }} változó az oldal címét jeleníti meg, míg a két <canvas> elem a sikeres és sikertelen DAG futásokat fogja ábrázolni. Az elemeken JavaScript segítségével ábrázoljuk a statisztikai adatokat, amelyeket a backend által megadott dag_run_stats lista tartalmaz.

További szempontok

A bemutatott megoldás alapvető, de számos lehetőség van annak bővítésére. Például a dashboard grafikus megjelenítését dinamikusabbá tehetjük JavaScript könyvtárakkal, mint például a Chart.js, hogy még több vizuális elemet adjunk hozzá. Ezen kívül érdemes lenne a rendszer teljesítményét monitorozni, különösen akkor, ha több ezer DAG futást kell követnünk. Ezen kívül a felhasználói jogkörök finomhangolása is elengedhetetlen, hogy a metrikák ne csak az arra jogosultak számára legyenek elérhetőek.

Hogyan készíthetünk egyéni Airflow szolgáltatót?

A szolgáltatók elkészítése előtt fontos megérteni az alapvető struktúrát, amely biztosítja a szolgáltató megfelelő működését az Airflow-ban. Az alábbiakban bemutatott mappa- és fájlszerkezet egy általánosan elfogadott és javasolt struktúrát követ, amelyet érdemes alkalmazni a közösségi szabványoknak megfelelően.

A fő mappastruktúra az alábbi elemeket tartalmazhatja:

1. Hooks mappa: Az itt található fájlok felelősek a külső szolgáltatásokkal való kapcsolat létrehozásáért, azaz a különféle horgok implementálásáért.

2. Operators mappa: Ez tartalmazza azokat az operátorokat, amelyek az Airflow DAG-okban (Directed Acyclic Graphs) használhatók. Minden operátornak rendelkeznie kell egy execute metódussal, amely felelős a végrehajtásért.

3. Sensors mappa: A sensorok olyan speciális operátorok, amelyek valamilyen külső eseményt figyelnek, és amíg az esemény nem következik be, folyamatosan várakoznak.

4. Triggers mappa: Speciális elhalasztott operátorokhoz használt triggerek találhatóak itt.

5. Dev mappa: Fejlesztési környezetet biztosít, például Docker-fájlokat és docker-compose.yaml-t tartalmaz, amelyek lehetővé teszik a szolgáltató helyi tesztelését és demonstrálását.

6. Example DAGs mappa: Itt találhatók a példákként szolgáló DAG-ok, amelyek bemutatják a szolgáltató működését.

7. Tests mappa: Tesztelés céljából találhatók benne a különféle egység- és integrációs tesztek.

A szolgáltató csomagjaival kapcsolatos további fontos fájlok a következők:

• setup.py és setup.cfg: Ezek tartalmazzák a csomag metaadatait és biztosítják, hogy a szolgáltató csomagja megfelelően telepíthető legyen.

• A setup.cfg fájl segítségével a szolgáltató információit, mint például a név, leírás és verzió, az Airflow számára hozzáférhetővé tesszük, hogy a felhasználói felületen, parancssori eszközökön és API-n keresztül láthatóvá váljon.

A szolgáltató kódjának írásakor fontos figyelembe venni, hogy az Airflow Scheduler folyamatosan elemzi a DAG-okat, és ha egy __init__ metódusban harmadik fél API-jait hívunk meg, azok a kód betöltődésekor lefutnak. Emiatt fontos, hogy az inicializáló metódusok ne tartalmazzanak olyan hívásokat, amelyek csak futás közben adnak vissza érvényes értékeket. Ehelyett Jinja sablonokat és makrókat kell használni a konfigurációk kezelésére a DAG-okban.

A szolgáltató regisztrálása során elengedhetetlen, hogy a csomagunk tartalmazza a megfelelő metainformációkat. Az apache_airflow_provider specifikációval regisztrálhatjuk a szolgáltatót az Airflow rendszerében. Ehhez létre kell hoznunk egy get_provider_info nevű függvényt, amely visszaad egy szótárat a csomag nevéről, leírásáról és verziójáról. A csomag telepítésekor az Airflow felismeri ezt a bejegyzést, és megjeleníti a szolgáltatót a felhasználói felületen.

Az operátorok működéséhez elengedhetetlen, hogy helyesen implementáljuk az egyes interfészeket. A példánkban egy TeaPotHook horgot készítünk, amely lehetővé teszi az Airflow számára, hogy egyszerű HTTP kérdésekkel kommunikáljon a teapottal. A horgon keresztül leírt kapcsolat részletezése biztosítja, hogy a külső rendszerrel való interakció során megfelelő adatok kerüljenek feldolgozásra.

A hook kódja a következő elemeket tartalmazza:

• conn_name_attr: A kapcsolati név attribútuma, amelyet az Airflow használ a kapcsolat azonosítására.

• get_connection_form_widgets: Ez a statikus metódus lehetővé teszi, hogy a kapcsolat adatainak megjelenítése az Airflow webes felületén történjen, ahol az egyes adatmezők testreszabhatók.

A megfelelő kapcsolati űrlapok és az Airflow webes felületének testreszabása segít a felhasználóknak a szolgáltató integrálásában, anélkül hogy bonyolult konfigurációval kellene szembenézniük.