Hvordan implementerer man en metrics-dashboard plugin i Flask?

Når vi arbejder med Airflow og Flask, kan vi skabe et plugin, der viser et metrics-dashboard til at overvåge DAG-kørsler og deres resultater. I denne proces opretter vi en Flask-baseret webgrænseflade, der trækker data fra Airflow-databasen og viser resultaterne i et dashboard med hjælpeværktøjer som Jinja-templating og Bootstrap for responsiv design.

Først skal vi oprette strukturen for vores plugin-modul. Denne struktur omfatter blandt andet en metrics_plugin-mappe, som vil indeholde alle nødvendige filer for både frontend og backend. I denne struktur har vi en __init__.py-fil, der registrerer både Flask-blueprintet og pluginet, samt en templates-mappe, der indeholder HTML-koden for frontend-dashbordet. Backend-koden, som håndterer dataforespørgslerne, placeres i views/dashboard.py.

Når det kommer til implementeringen af backend-koden, begynder vi med at definere de nødvendige imports i views/dashboard.py. Dette inkluderer moduler som AccessView, der sikrer, at kun autentificerede brugere har adgang til dashboardet, samt SQLAlchemy-sessionen, som muliggør forespørgsler på Airflow-databasen. Vi bruger BaseView fra Flask-AppBuilder som basis for vores egen MetricsDashboardView, hvor vi definerer vores rute og visning.

Følgende kode skaber en rute for vores dashboard, der vil vise metrics for DAG-kørsler:

python
Copy code
class MetricsDashboardView(BaseView):

    """En Flask-AppBuilder View for et metrics-dashboard"""
    default_view = "index"
    route_base = "/metrics_dashboard"

Her definerer vi MetricsDashboardView, og sætter nogle standardværdier for default_view og route_base. Denne struktur giver os mulighed for at oprette en brugervenlig rute under /metrics_dashboard.

For at vise metrics skal vi nu definere en funktion, der trækker data fra Airflow-databasen. Funktionen index henter information om succesfulde og fejlede DAG-kørsler for forskellige tidsperioder (1 dag, 7 dage og 30 dage):

python
Copy code
@provide_session
@expose("/")
@has_access_view(AccessView.PLUGINS)
def index(self, session: Session = NEW_SESSION):
    """Opret dashboard-visning"""
    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,
    )

I denne funktion bygger vi en SQL-forespørgsel, der trækker data om succesfulde og fejlede DAG-kørsler for de sidste 1, 7 og 30 dage. Resultatet af forespørgslen konverteres til en liste af ordbøger og sendes til frontend-skabelonen, som vi definerer næste gang.

HTML-koden til denne del kan se ud som følger:

html
Copy code
{% extends base_template %}
{% block title %}
    {{ title }}
{% endblock %}
{% block content %}
    <div class="container">
        <div class="row">
            <div class="col-md-6">

                <canvas id="successChart"></canvas>

            </div>
            <div class="col-md-6">
                <canvas id="failedChart"></canvas>
            </div>
        </div>
    </div>
{% endblock %}
{% block tail %}
    {{ super() }}
{% endblock %}

For at generere graferne, der viser success- og fejlslagne DAG-kørsler, skal vi bruge JavaScript. Dette skrives i tail-blokken af skabelonen, så vi kan opdatere grafikken dynamisk.

Udover selve koden er der flere vigtige overvejelser, som brugeren af dette plugin bør være opmærksom på. For det første skal sikkerheden være i fokus. Vi anvender has_access_view-dekorationen for at sikre, at kun brugere med de rette rettigheder kan få adgang til dashboardet. Desuden er det vigtigt at sikre, at forespørgslerne til databasen ikke bliver for tunge, især hvis systemet har mange DAG'er og kørselshistorik.

Slutteligt, når dashboardet er i brug, bør der være overvejelser om, hvordan man håndterer store mængder data. Flask og Airflow kan håndtere relativt store datamængder, men man bør altid være opmærksom på, hvordan systemet kan optimeres for både ydeevne og skalerbarhed.

Hvordan opretholder man effektiv testning og overvågning af Airflow-pipelines i en produktionsmiljø?

Testning af Airflow-pipelines og deres komponenter er en essentiel proces, der ikke kun garanterer kvaliteten af den implementerede kode, men også sikrer stabiliteten og pålideligheden af systemet i drift. For at opnå dette er det nødvendigt at tilpasse testmetoder og strategier til både den specifikke Airflow-opsætning og virksomhedens operationelle retningslinjer.

En vigtig tilgang er at opdele testene i forskellige niveauer og kontekster. Smøg- og enhedstests bør udføres i CI/CD-miljøet, hvor de hurtigst muligt kan identificere problemer i den skrevne kode uden at påføre yderligere systemkompleksitet. Functional tests kræver et QA-miljø, hvor testeren har friheden til at manipulere data for at simulere forskellige brugsscenarier. På den anden side kræver performance tests et staging-miljø, der er så tæt som muligt på produktionsmiljøet, for at simulere de reelle arbejdsforhold og evaluere systemets adfærd under belastning.

Når det kommer til Airflow-skripter og deres testning, er det essentielt at skelne mellem forskellige typer tests. For det første er smøgtestene meget simple og kontrollerer, om den basale funktionalitet virker, som f.eks. om Airflow-pakken kan installeres korrekt. Enhedstests er mere fokuserede og tester de mindste kodeenheder i isolation, uden afhængighed af eksterne tjenester. På den anden side tester funktionelle/integrationstests, hvordan koden interagerer med de eksterne systemer, der kræves af din Airflow-pipeline, og kontrollerer, at den fungerer som forventet i en produktionslignende kontekst.

For at maksimere effektiviteten af dine tests og workflows er det vigtigt at integrere CI/CD-processer i Airflow, så du kan automatisere og validere dine pipelines hurtigt. På den måde kan du undgå forsinkelser i udviklingen og sikre, at pipeline-strukturen og processerne fungerer fejlfrit i produktionen.

Scheduleren er kernen i Airflow og er ansvarlig for at planlægge og udføre opgaver. Det er afgørende at sikre, at schedulerens sundhed konstant overvåges for at undgå opgaver, der ikke bliver planlagt eller udført korrekt. En måde at gøre dette på er ved at overvåge metricen scheduler.scheduler_loop_duration, som giver indsigt i, hvor hurtigt schedulerens loop kører. Hvis værdien af denne metric stiger, kan det indikere, at schedulerens kapacitet ikke er tilstrækkelig til den arbejdsbyrde, der kræves. Yderligere målinger som scheduler.tasks.starving og scheduler.tasks.executable giver også værdifuld indsigt i, om systemet har nok ressourcer til at håndtere de opgaver, der er planlagt til udførelse.

Metadata-databasen er en anden kritisk komponent, da den gemmer vigtig information om DAGs, opgaver og deres status. Hvis der sker datatab i denne database, kan det resultere i, at opgaver bliver kørt flere gange, eller at systemet mister vigtige operationelle data. Derfor er det vigtigt at sikre, at metadata-databasen er korrekt overvåget, og at der er en katastrofegenopretningsplan på plads for at minimere eventuelle driftsforstyrrelser.

For at opsummere, når Airflow er implementeret i en produktionsmiljø, er det ikke kun selve kodens funktionalitet, der skal testes og monitoreres. Det er lige så vigtigt at forstå, hvordan systemet opfører sig i drift, og hvordan de forskellige komponenter arbejder sammen for at sikre en stabil og effektiv arbejdsflow. Overvågning af scheduler, metadata-database og andre kritiske komponenter i systemet er nødvendige for at sikre, at Airflow-pipelines fungerer optimalt og uden afbrydelser.

Hvordan planlægger man en migration mellem Airflow-miljøer?

Når Airflow-forekomsterne vokser, kan det blive nødvendigt at migrere arbejdsbelastninger til nye miljøer. Dette kan være nødvendigt af flere årsager – fra ændringer i tjenesteudbydere til behovet for at isolere arbejdsprocesser, der tidligere kørte i det samme miljø. Uanset årsagen er migration mellem Airflow-miljøer generelt en mere ligetil proces, og kræver som regel ikke så omfattende testning, som når man migrerer til et helt andet værktøj.

Før du går i gang med migrationen, er det vigtigt at identificere de DAGs, du ønsker at migrere, samt hvilke objekter der skal følge med for at sikre korrekt udførelse af arbejdsbelastningen. Det kan være nødvendigt at migrere tidligere kørselsdata, som historik og status, for at sikre en glidende overgang.

Forbindelser og variabler

For at få dine DAGs til at fungere korrekt i det nye miljø, skal du tage højde for eventuelle forbindelser og variabler, der er nødvendige for deres eksekvering. Hvis du bruger en secrets-backend til at håndtere variabler, skal du konsultere den service, du anvender, for at forstå, hvordan disse variabler migreres korrekt. Hvis du benytter miljøvariabler eller Kubernetes-hemmeligheder, bør du tjekke dokumentationen for din miljøvariabelmanager for at få vejledning om den bedste migrationsmetode.

Migrering af DAGs

Når de nødvendige forbindelser og variabler er migreret, og du har delt din kode i et nyt repository, er det næste skridt at færdiggøre migreringen og aktivere arbejdsbelastningerne i det nye miljø. Den simpleste metode er at deaktivere den oprindelige DAG, opdatere koden, så den nye startdato svarer til den sidste kørsel af DAG'en, og derefter aktivere den på den nye Airflow-instans. Dette fungerer godt, hvis du kun har et lille antal DAGs.

Men hvis din DAG ikke har en startdato, eller hvis den ikke kan ændres (f.eks. hvis "catchup" er aktiveret), kan du migrere kørselsdataene fra den gamle Airflow-instans til den nye. Dette gøres ved at bruge en forespørgsel på metadata-databasen, som henter de tidligere DAG-kørsler, og indsætter dem i den nye instans for at sikre, at begge miljøer har de samme operationelle tilstande.

python
Copy code
from airflow.models import DagRun

dag_runs = session.query(DagRun).filter(DagRun.dag_id == "dag_id")

Brug af pseudo-kode

Pseudo-koden, der er nævnt ovenfor, giver et hurtigt overblik over, hvordan du kan eksportere og importere data mellem to Airflow-instancer. Ved at identificere de nødvendige forbindelser og variabler kan du opnå en effektiv migration. Dette mønster kan også udvides til at inkludere andre objekter, der repræsenteres som kerne-datamodeller i Airflow, hvis nødvendigt.

For mindre migrationsprojekter, hvor antallet af forbindelser eller variabler er begrænset, kan du også vælge at migrere dem manuelt via Airflow UI. Dette er en praktisk metode, hvis du kun skal migrere få objekter.

Vigtige overvejelser under migration

Migrering af arbejdsflows og data mellem forskellige Airflow-miljøer er en aktivitet, som du sandsynligvis vil finde dig selv i flere gange i løbet af din karriere som dataingeniør. Når du planlægger og gennemfører migrationer, er det afgørende at afsætte tid til at planlægge grundigt før du går i gang, fremfor at forsøge at rette op på problemer efter migrationen. Jo bedre forberedelse, desto færre problemer vil opstå undervejs.

Desuden er det vigtigt at være opmærksom på, at mens migration mellem Airflow-miljøer ofte er enklere end migration til helt andre systemer, kræver det stadig en god forståelse af de underliggende Airflow-konfigurationer og metadata-strukturer. For eksempel, hvis du migrerer til et nyt miljø med en anden konfiguration, kan det være nødvendigt at justere ressourcer, såsom executor-konfigurationer eller operatørindstillinger, for at optimere ydeevnen i det nye miljø.