Generiranje dinamičkih nizova zadataka u protoku zraka pomoću konfiguracije DAG Run

Otključavanje snage ovisnosti o dinamičnim zadacima u protoku zraka

Apache Airflow je moćan alat za automatizaciju tijeka rada, ali rukovanje dinamičnim ovisnostima ponekad se može osjećati kao rješavanje slagalice. Prilikom dizajniranja usmjerenog acikličkog grafikona (DAG), sekvence zadataka tvrdog kodiranja mogu raditi za jednostavne slučajeve upotrebe, ali što ako strukturu treba utvrditi u vrijeme izvođenja? 🤔

Zamislite da radite na cjevovodu podataka gdje zadaci koji se izvršavaju ovise o dolaznim podacima. Na primjer, obrada različitih skupova datoteka na temelju dnevne konfiguracije ili izvršavanja varijabilnih transformacija na temelju poslovnog pravila. U takvim slučajevima, statički DAG neće ga smanjiti - potreban vam je način da dinamički definirate ovisnosti.

Upravo je to mjesto gdje je protok zraka dag_run.conf može biti izmjenjivač igre. Prolazeći konfiguracijski rječnik prilikom pokretanja DAG -a, možete dinamički generirati sekvence zadataka. Međutim, to implementiranje na strukturirani način zahtijeva duboko razumijevanje modela izvršenja zraka.

U ovom ćemo članku istražiti kako izgraditi dinamični DAG u kojem se određuju ovisnosti o zadacima tijekom izvođenja koristeći dag_run.conf. Ako ste se borili da to postignete i niste pronašli jasno rješenje, ne brinite - niste sami! Razdvojimo ga korak po korak s praktičnim primjerima. 🚀

Izgradnja dinamičnih bodova s ​​konfiguracijom izvođenja u protoku zraka

Apache Airflow je moćan alat za orkestriranje složenih tijekova rada, ali njegova istinska snaga leži u njegovoj fleksibilnosti. Skripte predstavljene ranije pokazuju kako stvoriti a dinamički DAG gdje se ovisnosti o zadacima određuju tijekom izvođenja koristeći dag_run.conf. Umjesto tvrdog kodiranja popisa elemenata za obradu, DAG ih dinamički dohvaća kad se pokrene, omogućujući prilagodljiviji tijek rada. To je posebno korisno u stvarnim scenarijima, poput obrade varijabilnih skupova podataka ili izvršavanja određenih zadataka na temelju vanjskih uvjeta. Zamislite ETL cjevovod u kojem datoteke za obradu svakodnevno mijenjaju - ovaj pristup čini automatizaciju znatno lakšim. 🚀

Prva skripta koristi Pitonoperator izvršavati zadatke i dinamički postaviti ovisnosti. Izdvaja popis elemenata iz dag_run.conf, osiguravajući da se zadaci stvaraju samo kad je potrebno. Svaki element na popisu postaje jedinstven zadatak, a ovisnosti se postavljaju uzastopno. Drugi pristup koristi API API -a, što pojednostavljuje stvaranje DAG -a s dekoratorima poput @dag i @zadatak. Ova metoda čini DAG čitljivijom i održava čišće logiku izvršenja. Ovi pristupi osiguravaju da se tijekovi rada mogu prilagoditi različitim konfiguracijama bez potrebe za promjenama koda.

Na primjer, razmotrite scenarij u kojem tvrtka za e-trgovinu obrađuje narudžbe u serijama. Neki dan mogu imati hitnije naredbe od drugih, što zahtijeva različite sekvence zadataka. Korištenje statičkog DAG značilo bi izmjenu koda svaki put kada se prioriteti promijene. Pomoću našeg dinamičnog DAG pristupa, vanjski sustav može pokrenuti DAG specifičnim redoslijedom zadatka, što postupak čini učinkovitijim. Drugi slučaj upotrebe je u znanosti o podacima, gdje će modeli možda trebati prekvalifikaciju na temelju dolaznih raspodjela podataka. Prolazeći potrebnim konfiguracijama modela dinamički, izvršavaju se samo potrebni proračuni, štedeći vrijeme i resurse. 🎯

Ukratko, ove skripte pružaju temelj za dinamički generiranje DAG -ova na temelju unosa izvođenja. Iskorištavanjem API API AIRFOW -a Ili tradicionalni pristup pitonoperatora, programeri mogu stvoriti fleksibilne, modularne i učinkovite tokove rada. To eliminira potrebu za ručnom intervencijom i omogućava bešavnu integraciju s drugim sustavima automatizacije. Bez obzira na obradu narudžbi kupaca, upravljanje cjevovodima za podatke ili orkestriranje tijekova rada u oblaku, dinamični DAG -ovi omogućuju pametniju automatizaciju prilagođenu specifičnim poslovnim potrebama.

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

Poboljšanje sekvenciranja dinamičkog zadatka s uvjetnim izvršavanjem u protoku zraka

Jedna moćna, ali često previdjena značajka u Apache protok zraka je uvjetno izvršenje, što može dodatno poboljšati fleksibilnost dinamičkog sekvenciranja zadataka. Dok je dohvaćao ovisnosti o zadacima iz dag_run.conf Korisni je, u stvarnom scenariji često zahtijevaju izvršavanje samo određenih zadataka na temelju određenih uvjeta. Na primjer, neki skupovi podataka mogu zahtijevati unaprijed obradu prije analize, dok se drugi mogu izravno obraditi.

Uvjetno izvršenje u protoku zraka može se provesti pomoću BranchPythonOperator, koji određuje sljedeći zadatak izvršiti na temelju unaprijed definirane logike. Pretpostavimo da imamo dinamični DAG koji obrađuje datoteke, ali samo datoteke iznad određene veličine zahtijevaju provjeru valjanosti. Umjesto da sve zadatke izvršavaju uzastopno, dinamički možemo odlučiti koje će zadatke izvršiti, optimizirati vrijeme izvršenja i smanjenje upotrebe resursa. Ovaj pristup osigurava pokretanje samo relevantnih tijekova rada, što čini cjevovode podataka učinkovitijim. 🚀

Drugi način poboljšanja dinamičnih DAG -a je uključivanje XComs (Poruke unakrsnih komunikacija). XCOM -ovi dopuštaju zadacima da razmjenjuju podatke, što znači da dinamički stvoreni redoslijed zadatka može prenijeti informacije između koraka. Na primjer, u ETL cjevovodu zadatak prethodne obrade može odrediti potrebne transformacije i te detalje proslijediti na sljedeće zadatke. Ova metoda omogućuje uistinu tijekove rada na temelju podataka, gdje se protok izvršenja prilagođava na temelju ulaza u stvarnom vremenu, značajno povećavajući mogućnosti automatizacije.