Dünaamiliste ülesannete järjestuste genereerimine Airflow'is DAG Run konfiguratsiooni abil

Dünaamiliste ülesannete sõltuvuste võimu avamine õhuvoolus

Apache Airflow on võimas töövoo automatiseerimisriist, kuid dünaamiliste sõltuvuste haldamine võib mõnikord tunda pusle lahendamist. Juhtneva atsüklilise graafiku (DAG) kavandamisel võivad kõva kodeerivad ülesandejärjestused töötada lihtsate kasutusjuhtumite jaoks, kuid mis siis, kui struktuur tuleb käitumisel kindlaks teha? 🤔

Kujutage ette, et töötate andmetorustikul, kus täidetavad ülesanded sõltuvad sissetulevatest andmetest. Näiteks töötlemine erinevate failide komplektide töötlemine põhineb igapäevasel konfiguratsioonil või käivitades muutuva teisenduse, mis põhineb ärireeglil. Sellistel juhtudel ei lõika staatiline DAG seda - vajate viisi, kuidas sõltuvusi dünaamiliselt määratleda.

Just see on see, kus Airflow on dag_run.conf võib olla mängude vahetaja. DAG -i käivitamisel konfiguratsioonisõnastiku edastades saate dünaamiliselt genereerida ülesandejärjestusi. Selle struktureeritud viisil rakendamine nõuab aga AirFlowi täitmismudeli sügavat mõistmist.

Selles artiklis uurime, kuidas ehitada dünaamiline DAG, kus käitusajal määratakse ülesande sõltuvused dag_run.conf. Kui olete selle saavutamise nimel vaeva näinud ja pole leidnud selget lahendust, ärge muretsege - te pole üksi! Jagagem see samm -sammult praktiliste näidetega. 🚀

Dünaamiliste DAG -de ehitamine tööaja konfiguratsiooniga Airflow

Apache AirFlow on võimas tööriist keerukate töövoogude korraldamiseks, kuid selle tegelik tugevus seisneb selle paindlikkuses. Varem esitatud skriptid näitavad, kuidas luua a dünaamiline dag kus ülesande sõltuvused määratakse käitusajal, kasutades dag_run.conf. Töötletavate elementide loendi raskesti kodeerimise asemel hangib DAG need käivitamisel dünaamiliselt, võimaldades kohanemisvõimelisemaid töövooge. See on eriti kasulik reaalainete stsenaariumide korral, näiteks muutuvate andmekogumite töötlemine või väliste tingimuste põhjal konkreetsete ülesannete täitmine. Kujutage ette ETL -torujuhtme, kus failid töötlevad iga päev - see lähenemisviis muudab automatiseerimise palju lihtsamaks. 🚀

Esimeses skriptis kasutatakse Pütonoperaator ülesannete täitmiseks ja sõltuvuste dünaamiliseks määramiseks. See eraldab elementide loendist dag_run.conf, tagades ülesannete loomise ainult vajadusel. Iga loendi element muutub ainulaadseks ülesandeks ja sõltuvused seatakse järjestikku. Teine lähenemisviis kasutab Taskflow API, mis lihtsustab DAG -i loomist dekoraatoritega @DAG ja @task. See meetod muudab DAG loetavamaks ja säilitab puhtama täitmisloogika. Need lähenemisviisid tagavad, et töövood saaksid erinevate konfiguratsioonidega kohaneda ilma koodimuudatusi nõudmata.

Näiteks kaaluge stsenaariumi, kus e-kaubandusettevõte töötleb tellimusi partiidena. Mõnel päeval võib olla rohkem kiireloomulisi tellimusi kui teistel, nõudes erinevaid ülesandejärjestusi. Staatilise DAG kasutamine tähendaks koodi muutmist iga kord, kui prioriteedid muutuvad. Meie dünaamilise DAG -lähenemisviisi abil saab väline süsteem käivitada DAG konkreetse ülesandejärjestusega, muutes protsessi tõhusamaks. Teine kasutusjuhtum on andmeteadus, kus mudelid võivad vajada sissetulevate andmejaotuste põhjal ümberõppet. Vajalike mudeli konfiguratsioonide dünaamiliselt edastades teostatakse ainult vajalikke arvutusi, säästes aega ja ressursse. 🎯

Kokkuvõtlikult annavad need skriptid aluse DAG -de dünaamiliseks genereerimiseks, mis põhinevad tööaja sisenditel. Võimendades Airflowi Taskflow API Või traditsioonilise pütonoperaatorite lähenemisviisi, arendajad saavad luua paindlikke, modulaarseid ja tõhusaid töövooge. See välistab käsitsi sekkumise vajaduse ja võimaldab sujuvat integreerimist teiste automatiseerimissüsteemidega. Ükskõik, kas töötlemine klientide tellimuste töötlemine, andmete torujuhtmete haldamine või pilve töövoogude korraldamine, dünaamilised DAG -id võimaldavad nutikamat automatiseerimist, mis on kohandatud konkreetsetele ärivajadustele.

from airflow import DAG
from airflow.operators.python import PythonOperator
from airflow.utils.dates import days_ago
from airflow.models import DagRun
import json
# Define default args
default_args = {
    'owner': 'airflow',
    'depends_on_past': False,
    'start_date': days_ago(1),
}
# Function to process each element
def process_element(element, kwargs):
    print(f"Processing element: {element}")
# Define DAG
dag = DAG(
    'dynamic_task_dag',
    default_args=default_args,
    schedule_interval=None,
)
# Extract elements from dag_run.conf
def generate_tasks(kwargs):
    conf = kwargs.get('dag_run').conf or {}
    elements = conf.get('elements', [])
    task_list = []
    for i, group in enumerate(elements):
        for j, element in enumerate(group):
            task_id = f"process_element_{i}_{j}"
            task = PythonOperator(
                task_id=task_id,
                python_callable=process_element,
                op_args=[element],
                dag=dag,
            )
            task_list.append(task)
    return task_list
# Generate dynamic tasks
tasks = generate_tasks()
# Define dependencies dynamically
for i in range(len(tasks) - 1):
    tasks[i + 1].set_upstream(tasks[i])

from airflow.decorators import dag, task
from datetime import datetime
# Define DAG
@dag(schedule_interval=None, start_date=datetime(2025, 1, 28), catchup=False)
def dynamic_taskflow_dag():
    @task
    def process_element(element: str):
        print(f"Processing {element}")
    @task
    def extract_elements(dag_run=None):
        conf = dag_run.conf or {}
        return conf.get('elements', [])
    elements = extract_elements()
    task_groups = [[process_element(element) for element in group] for group in elements]
    # Define dependencies dynamically
    for i in range(len(task_groups) - 1):
        for upstream_task in task_groups[i]:
            for downstream_task in task_groups[i + 1]:
                downstream_task.set_upstream(upstream_task)
dynamic_taskflow_dag()

Dünaamilise ülesande sekveneerimise parandamine koos tingimusliku täitmisega õhuvoolul

Üks võimas, kuid sageli tähelepanuta jäetud funktsioon Apache õhuvool on tingimuslik täitmine, mis võib veelgi parandada ülesande dünaamilise järjestamise paindlikkust. Ülesande sõltuvuste hankimise ajal dag_run.conf on kasulik, reaalmaailma stsenaariumid nõuavad sageli ainult teatud tingimuste põhjal teatud ülesannete täitmist. Näiteks võivad mõned andmekogumid vajada enne analüüsi eeltöötlemist, teisi aga otse töödelda.

Tingimuslikku täitmist õhuvooses saab rakendada kasutades BranchPythonOperator, mis määrab eelnevalt määratletud loogika põhjal järgmise ülesande. Oletame, et meil on dünaamiline DAG, mis töötleb faile, kuid ainult teatud suurusest kõrgemad failid nõuavad valideerimist. Kõigi ülesannete järjestikuse täitmise asemel saame dünaamiliselt otsustada, milliseid ülesandeid käivitada, optimeerides täitmisaega ja vähendada ressursside kasutamist. See lähenemisviis tagab, et käivitatakse ainult asjakohased töövood, muutes andmetorustikud tõhusamaks. 🚀

Teine viis dünaamiliste DAG -de täiustamiseks on kaasamine XComs (Ristkommunikatsioonisõnumid). XCOMS võimaldab ülesandeid vahetada andmeid, mis tähendab, et dünaamiliselt loodud ülesandejärjestus võib teavet edastada sammude vahel. Näiteks võib ETL -torujuhtmes eeltöötlusülesanne kindlaks määrata nõutavad teisendused ja edastada need üksikasjad järgmistele ülesannetele. See meetod võimaldab tõeliselt andmepõhiseid töövooge, kus täitmisvool kohandab reaalajas sisendite põhjal, suurendades automatiseerimisvõimalusi märkimisväärselt.