Hvordan automatisere din kode med en Airflow DAG?

Automatisering af kode med en Airflow DAG kræver, at vi definerer nogle vigtige elementer og bruger en række værktøjer, som vi endnu ikke har dækket. Et af de vigtigste skridt er at forstå, hvordan Airflow fungerer, og hvordan det kan bruges til at strukturere og automatisere arbejdsprocesser effektivt. Airflow DAG (Directed Acyclic Graph) er en metode til at organisere og planlægge opgaver, så de køres på bestemte tidspunkter eller med bestemte afhængigheder. I denne sektion ser vi på, hvordan du kan bruge Airflow til at automatisere en simpel opgave, såsom at hente dagens billede fra NASA's API, og derefter sende en meddelelse om, at billedet er blevet tilføjet.

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)

Nu skal vi definere de standardargumenter, der bruges til at konfigurere DAG'en. Dette giver Airflow de nødvendige parametre til at udføre opgaverne korrekt:

python
Copy code
default_args = {
    'owner': dag_owner,
    'depends_on_past': False,
    'retries': 2,
    'retry_delay': timedelta(minutes=5)
}

Med disse indstillinger er vi klar til at initialisere DAG'en ved hjælp af with-sætningen, hvilket forhindrer Airflow scheduler fra at initialisere DAG-objekter hver gang den parser koden:

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']
):

Herefter definerer vi get_pictures-tasken ved hjælp af en PythonOperator og angiver den funktion, vi definerede tidligere som python_callable:

python
Copy code
get_pictures = PythonOperator(
    task_id="get_pictures",
    python_callable=_get_pictures,
)

Vi tilføjer også en notify-task for at informere os, når billedet er blevet tilføjet. For enkelhedens skyld bruger vi en BashOperator, men dette kan let ændres til en anden type operator, der sender en meddelelse via for eksempel e-mail eller Slack:

python
Copy code
notify = BashOperator(

    task_id="notify",
    bash_command='echo "Image for today has been added!"',
)

Endelig definerer vi task execution path ved at sætte get_pictures som den første opgave og notify som den næste opgave:

python
Copy code
get_pictures >> notify

Skrive din første DAG

Koden til at skrive en DAG kan virke mere kompleks end koden i et Jupyter Notebook, men meget af den gentages ofte. I DAG-koden findes flere standardfelter, der hjælper Airflow med at forstå, hvordan det skal fungere. For en dataingeniør er det vigtigt at forstå, at det er muligt at opdele manuelle og højvolumenopgaver i flere automatiserede opgaver, som alle danner en DAG. I eksemplet her bliver opgaverne udført i rækkefølge, men i mere komplekse scenarier kan Airflow køre opgaverne parallelt.

Når vi opretter en DAG, er det vigtigt at identificere, hvilken DAG der hører til hvilke opgaver. Her skal du forstå, at DAG'ens navn og dens attributter identificeres, så Airflow kan forstå, hvad der hører til hvilken opgave.

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']
):

En beskrivelse af DAG'en er altid en god idé, da det giver udvikleren mulighed for at forklare, hvad denne DAG skal opnå. Dette er især nyttigt, når flere udviklere arbejder sammen, og det hjælper med at holde koden forståelig. Startdatoen fastsætter den oprindelige dato, hvor DAG'en skal begynde at køre sine opgaver. Hvis vi for eksempel ønsker, at den skal starte i dag, kan vi sætte startdatoen til 0.

Felter som catchup er vigtige, da de bestemmer, om Airflow skal køre opgaver for alle de perioder, der mangler, hvis startdatoen er længere tilbage. Hvis catchup er sat til True, og vi har sat startdatoen til 7 dage siden, vil Airflow forsøge at køre alle opgaver for de 7 dage. I de fleste tilfælde, når vi starter en DAG, vil vi ikke nødvendigvis bruge catchup, især hvis vi kun ønsker at køre opgaverne fremadrettet.

Til sidst har vi tags, som hjælper med at organisere DAG'erne. Dette gør det muligt at kategorisere og filtrere DAG'erne, hvilket kan være nyttigt, når du arbejder med mange forskellige workflows.

Vigtige betragtninger

Når du arbejder med Airflow, skal du huske, at du kan have flere opgaver, der kører parallelt, eller opgaver, der afhænger af andre opgaver. For mere komplekse arbejdsflows, vil du kunne bygge opgaver, der arbejder uafhængigt af hinanden, men det kræver omhyggelig planlægning af afhængigheder og task rækkefølge.

Desuden bør du være opmærksom på Airflow’s evne til at håndtere fejl og genkørsel. Hvis en opgave fejler, kan Airflow automatisk forsøge at køre opgaven igen. For at optimere systemets pålidelighed skal du angive passende retry-mekanismer og fejlmeddelelser.

Når du skriver DAG-koden, er det også vigtigt at gøre den så forståelig som muligt for andre udviklere, så de kan vedligeholde og videreudvikle workflows uden problemer.

Hvordan vælge den rigtige repository-struktur for dit Airflow-projekt

Repository-struktur er en beslutning, der kan virke triviel, men som har stor betydning for effektiviteten og skalerbarheden af dit projekt. Når du vælger, hvordan dit repositoriesystem skal organiseres, er det vigtigt at tage højde for, hvordan dit team arbejder, hvilke operationelle mønstre der skal understøttes, og hvordan teamet vil interagere med systemet. Det er en beslutning, der kan ændres senere, men det er stadig noget, der kræver omhyggelig overvejelse.

På den anden side findes multi-repo-strukturen, hvor du opdeler hele dit Airflow-setup i flere separate repositories. For eksempel kunne du have et repo for Airflow's kerne, et repo for hver provider og plugin, samt et repo for hver enkelt teams DAGs. Denne struktur gør det muligt for teams at arbejde i et isoleret og simpelt workflow, hvilket kan være en fordel i større organisationer, hvor forskellige teams arbejder uafhængigt. Ulempen ved denne struktur er, at der kan opstå synkroniseringsproblemer, når ændringer skal trækkes sammen på tværs af alle systemer. Dette kræver typisk intensiv integrationstestning før frigivelse.

Valget mellem mono-repo og multi-repo er ikke enten-eller. De er blot yderpunkter på en skala, og det er sandsynligt, at din organisation ender med at bruge en blanding af de to. For eksempel kan du vælge en mono-repo til din Airflow-opsætning og de tilhørende interne providers, mens du opretter separate repositories for DAGs, som teams kan arbejde på. Denne tilgang gør det lettere for teams at arbejde lokalt og udvikle deres DAGs, samtidig med at de kan integreres med den centrale Airflow-opsætning.

Når du har valgt din repository-struktur, skal du overveje, hvordan du vil håndtere forbindelses- og variable-objekter i dit Airflow-setup. Disse objekter er essentielle for at definere, hvordan Airflow kommunikerer med de systemer, det interagerer med, og hvordan de forskellige DAGs skal fungere afhængigt af de værdier, der er defineret i variablerne. Da disse objekter ofte indeholder følsomme informationer, er det afgørende at vælge en metode til at håndtere dem, som er sikker og skalerbar.

For at sikre en korrekt håndtering af konfigurationer på tværs af forskellige miljøer, kan du bruge miljøvariabler til at definere forbindelser og variabler. Denne metode integrerer godt med moderne DevOps-praksisser og muliggør en mere automatiseret og sikker måde at håndtere følsomme data på. Eksempelvis kan du definere dine forbindelser og variabler som miljøvariabler, hvilket sikrer, at de tilgås korrekt, når Airflow starter op. Det er dog vigtigt at bemærke, at dette ikke nødvendigvis er den mest sikre metode, især hvis disse variabler lagres som rå tekststrenge. For at beskytte dataene kan du anvende et eksternt hemmelighedshåndteringssystem, som AWS Secrets Manager eller HashiCorp Vault, til at gemme disse oplysninger.

Når du har taget beslutninger om repository-struktur og sikkerhed, er næste skridt at vælge den rette metode til at implementere og distribuere dit Airflow-miljø. Afhængigt af dine organisationers infrastruktur og kapabiliteter, kan du vælge en metode, der passer bedst til dine ressourcer. Kubernetes er en populær mulighed, da det tilbyder en effektiv måde at køre komplekse applikationer som Airflow. Det giver en simpel deklarativ syntaks til at beskrive din applikation og håndterer de nødvendige afhængigheder mellem de forskellige services. Hvis du allerede har erfaring med Kubernetes, kan det være den bedste løsning for dig. Alternativt kan du vælge at bruge virtuelle maskiner eller direkte hardware (bare metal), hvis du har specifikke krav eller ønsker mere kontrol over din infrastruktur.

Valget af implementeringsmetode afhænger ikke kun af tekniske krav, men også af organisatoriske faktorer som budgetter og interne kompetencer. Det er vigtigt at tage højde for den viden og de ressourcer, der er tilgængelige, når du træffer beslutningen.

Hvordan Airflow kan optimere arbejdsgange i din organisation

Et af de mest fremtrædende anvendelsesområder er ETL-pipelines for data. Stort set enhver implementering af Airflow hjælper med at automatisere opgaver af denne type, hvad enten det drejer sig om at konsolidere data i et datalager eller flytte data mellem forskellige værktøjer. Dette er særligt nyttigt i store organisationer, hvor data kontinuerligt skal strømme mellem forskellige systemer og applikationer.

En anden vigtig brugssag er udvikling og distribution af brugerdefinerede plugins for organisationer med et unikt stack og behov, der ikke er blevet adresseret af open source-samfundet. Airflow giver mulighed for nem tilpasning af miljøet og økosystemet til specifikke behov. Dette gør det muligt for organisationer at udvide funktionaliteten af Airflow og integrere det med deres egne systemer og værktøjer, hvilket gør det endnu mere fleksibelt og tilpasset specifikke krav.

En fjerde anvendelse af Airflow er i ML-workflows og orkestrering på tværs af systemer. Mange teams, der bygger og vedligeholder ML-workflows, er afhængige af Airflow til at automatisere træning, transformation og evaluering af modeller. I disse scenarier hjælper Airflow med at sikre, at processerne forbliver strømlinede og kan håndtere den komplekse afhængighed mellem forskellige opgaver i ML-pipelines.

Zen of Python i forbindelse med Airflow

Når man arbejder med Airflow, er det essentielt at følge nogle grundlæggende principper, der guider udviklingen af kode. Et af de bedste udgangspunkt er "Zen of Python", et sæt af 19 principper skrevet af Tim Peters, som hjælper udviklere med at skrive mere effektiv og vedligeholdelsesvenlig kode. Selvom Zen of Python blev skrevet med Python-programmering i tankerne, er flere af principperne særligt nyttige, når man arbejder med Airflow.

En af de centrale ideer i Zen of Python er, at "det eksplicitte er bedre end det implicitte." Når man arbejder med dataopgaver, er det vigtigt at være eksplicit om sine intentioner, da dette letter både fejlfinding og vedligeholdelse. I en kompleks data-pipeline vil det være lettere at spore problemer, hvis man tydeligt angiver, hvad hver del af systemet skal gøre.

Idempotens og kode som konfiguration

Idempotens er et andet kritisk begreb, der bør overvejes under designfasen af Airflow-opgaver. Idempotens beskriver en operation, der kan udføres flere gange uden at ændre resultatet efter første udførelse. Ved at sikre, at systemet er idempotent, kan man undgå nogle af de mest almindelige driftsproblemer, da det garanterer, at gentagne kørsel af en opgave ikke skaber utilsigtede bivirkninger.

"Kode som konfiguration" er et designprincip, der gør det muligt at definere konfigurationsparametre som kode, adskilt fra den udførende kodebase. Denne tilgang giver mulighed for bedre versionstyring og sikrer, at ændringer i systemets konfiguration kan integreres og administreres på en struktureret måde. Dog er det vigtigt at være opmærksom på, at dette designprincip kan føre til utilsigtede afhængigheder, hvis konfigurationen ikke er klart defineret i kodebasen.

De vigtigste færdigheder til effektiv brug af Airflow

For at udnytte Airflow effektivt er det nødvendigt med flere færdigheder udover blot kendskab til værktøjet. At kunne skrive og forstå Python-kode er en grundlæggende færdighed, da både DAG'er og plugins skrives i Python. Det er også vigtigt, at organisationen har etableret solide testpraksisser, så det er muligt at forvente, at de udviklede workflows og plugins fungerer som tilsigtet. Automatiserede tests bør køres, før workflows implementeres i produktionsmiljøet.

Endelig er det nødvendigt at have kompetencer inden for applikationsmonitorering og alarmering. Når du arbejder med komplekse arbejdsprocesser og store datamængder, er det essentielt at kunne overvåge systemets sundhed og sende alarmer ved fejl eller kritiske hændelser. Dette sikrer, at der hurtigt kan gribes ind, hvis der opstår problemer, og at systemet forbliver stabilt og effektivt.