Dinaminių užduočių sekų generavimas oro sraute naudojant DAG paleidimo konfigūraciją

Oro srauto dinaminių užduočių priklausomybių galios atrakinimas

„Apache AirFlow“ yra galingas darbo eigos automatizavimo įrankis, tačiau dinaminių priklausomybių tvarkymas kartais gali jaustis kaip išspręsti galvosūkį. Projektuojant nukreiptą aciklinį grafiką (DAG), kietos kodavimo užduočių sekos gali veikti paprastais naudojimo atvejais, tačiau kas būtų, jei struktūrą reikia nustatyti vykdant? 🤔

Įsivaizduokite, kad dirbate prie duomenų vamzdyno, kuriame vykdomos užduotys priklauso nuo gaunamų duomenų. Pvz., Įvairių failų rinkinių apdorojimas, remiantis dienos konfigūracija ar kintamų transformacijų vykdymu, remiantis verslo taisykle. Tokiais atvejais statinis DAG jo nepaliks - jums reikia būdo, kaip dinamiškai apibrėžti priklausomybes.

Būtent ten oro srautas dag_run.conf Gali būti žaidimų keitiklis. Pasinaudodami konfigūracijos žodynu, kai suaktyvinsite DAG, galite dinamiškai generuoti užduočių sekas. Tačiau norint tai įgyvendinti struktūrizuotu būdu, reikia gerai suprasti „Airflow“ vykdymo modelį.

Šiame straipsnyje mes ištirsime, kaip sukurti dinaminį DAG dag_run.conf. Jei stengiatės to pasiekti ir neradote aiškaus sprendimo, nesijaudinkite - jūs ne vienas! Pažymėkime žingsnis po žingsnio su praktiniais pavyzdžiais. 🚀

Dinaminių DAG kūrimas su vykdymo laiko konfigūracija oro sraute

„Apache Airflow“ yra galingas įrankis, skirtas organizuoti sudėtingas darbo eigas, tačiau tikroji jo stiprumas slypi jo lankstume. Anksčiau pateikti scenarijai parodo, kaip sukurti a Dinaminis Dagas kur priklausomybės užduoties nustatomos vykdant laiką naudojant dag_run.conf. Užuot sunkiai kodavęs elementų, kuriuos reikia apdoroti, sąrašą, DAG juos dinamiškai nuskaito, kai suaktyvina, leisdamas labiau pritaikyti darbo eigas. Tai ypač naudinga realaus pasaulio scenarijuose, tokiuose kaip kintamųjų duomenų rinkinių apdorojimas arba konkrečios užduotys, pagrįstos išorinėmis sąlygomis, vykdant konkrečias užduotis. Įsivaizduokite ETL vamzdyną, kuriame failai, skirti apdoroti keičiasi kasdien - šis požiūris palengvina automatizavimą. 🚀

Pirmasis scenarijus naudoja Pythonopoperator Norėdami dinamiškai atlikti užduotis ir nustatyti priklausomybes. Jis ištraukia elementų sąrašą iš dag_run.conf, užtikrinant, kad užduotys būtų sukurtos tik tada, kai reikia. Kiekvienas sąrašo elementas tampa unikalia užduotimi, o priklausomybės nustatomos paeiliui. Antrasis požiūris pasinaudoja „TaskFlow API“, tai supaprastina DAG kūrimą su dekoratoriais, pavyzdžiui, @Dag ir @Task. Šis metodas daro DAG lengvesnį ir palaiko švaresnės vykdymo logiką. Šie metodai užtikrina, kad darbo srautai gali prisitaikyti prie skirtingų konfigūracijų, nereikalaujant kodo pakeitimų.

Pvz., Apsvarstykite scenarijų, kai elektroninės komercijos įmonė apdoroja užsakymus partijose. Kai kuriomis dienomis gali būti skubesni užsakymai nei kitos, reikalaujant skirtingų užduočių sekų. Statinio DAG naudojimas reikštų kodo keitimą kiekvieną kartą prioritetų keitimą. Taikant mūsų dinaminį DAG metodą, išorinė sistema gali suaktyvinti DAG specifinę užduočių seką, todėl procesas tampa efektyvesnis. Kitas naudojimo atvejis yra duomenų moksle, kai modeliams gali reikėti perkvalifikuoti remiantis gaunamų duomenų paskirstymais. Dinamiškai perduodant reikiamas modelio konfigūracijas, vykdomi tik būtini skaičiavimai, taupant laiką ir išteklius. 🎯

Apibendrinant galima pasakyti, kad šie scenarijai suteikia pagrindą dinamiškai generuoti DAG, pagrįstą vykdymo laiko įvestimis. Pasinaudodamas „AirFlow“ užduočių srauto API Arba tradicinis „Pythonoperator“ požiūris, kūrėjai gali sukurti lankstus, modulinius ir efektyvius darbo eigas. Tai pašalina rankinio intervencijos poreikį ir leidžia sklandžiai integruoti su kitomis automatikos sistemomis. Nesvarbu, ar apdorojate klientų užsakymus, tvarkant duomenų vamzdynus ar organizuojant debesų darbo eigas, „Dynamic DAG“ įgalina protingesnę automatizavimą, pritaikytą konkretiems verslo poreikiams.

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()

Dinaminės užduočių sekos sustiprinimas naudojant sąlyginį vykdymą oro sraute

Viena galinga, tačiau dažnai nepastebima funkcija „Apache AirFlow“ yra sąlyginis vykdymas, kuris gali dar labiau pagerinti dinaminių užduočių sekos lankstumą. Tuo tarpu gaunant priklausomybes nuo užduoties dag_run.conf yra naudingi, realaus pasaulio scenarijai dažnai reikia atlikti tik tam tikras užduotis, pagrįstas konkrečiomis sąlygomis. Pavyzdžiui, kai kuriems duomenų rinkiniams prieš analizę gali prireikti iš anksto apdoroti, o kiti gali būti tiesiogiai apdorojami.

Sąlyginį vykdymą oro sraute galima įgyvendinti naudojant BranchPythonOperator, kuris nustato kitą užduotį, kurią reikia vykdyti pagal iš anksto nustatytą logiką. Tarkime, kad turime dinaminį DAG, kuris apdoroja failus, tačiau reikia patvirtinti tik tam tikro dydžio failus. Užuot vykdę visas užduotis iš eilės, galime dinamiškai nuspręsti, kurias užduotis reikia vykdyti, optimizuoti vykdymo laiką ir sumažinti išteklių naudojimą. Šis metodas užtikrina, kad suaktyvinamos tik svarbios darbo eigos, todėl duomenų vamzdynai tampa efektyvesni. 🚀

Kitas būdas pagerinti dinaminius DAG yra įtraukiant XComs (Kryžminio komunikacijos pranešimai). „XCOMS“ leidžia užduotis keistis duomenimis, tai reiškia, kad dinamiškai sukurta užduočių seka gali perduoti informaciją tarp žingsnių. Pvz., ETL vamzdyne išankstinio apdorojimo užduotis gali nustatyti reikiamas transformacijas ir perduoti šias detales į vėlesnes užduotis. Šis metodas įgalina iš tikrųjų duomenų pagrįstus darbo eigas, kai vykdymo srautas prisitaiko atsižvelgiant į realaus laiko įvestis, žymiai didinančias automatizavimo galimybes.