Dinamisku uzdevumu secību ģenerēšana gaisa plūsmā, izmantojot DAG palaišanas konfigurāciju

Dinamisko uzdevumu atkarību atrašana gaisa plūsmā

Apache Airflow ir jaudīgs darbplūsmas automatizācijas rīks, taču dinamisko atkarību apstrāde dažreiz var justies kā mīklas risināšana. Izstrādājot virzītu aciklisko grafiku (DAG), hardkodēšanas uzdevumu sekvences varētu darboties vienkāršās lietošanas gadījumos, bet kā būtu, ja struktūra jānosaka izpildlaikā? 🤔

Iedomājieties, ka jūs strādājat pie datu cauruļvada, kurā izpildāmie uzdevumi ir atkarīgi no ienākošajiem datiem. Piemēram, dažādu failu kopu apstrāde, pamatojoties uz ikdienas konfigurāciju vai mainīgu pārveidošanas izpildi, pamatojoties uz biznesa noteikumu. Šādos gadījumos statisks DAG to nesagriezīs - jums ir nepieciešams veids, kā dinamiski definēt atkarības.

Tas ir tieši tad, ja gaisa plūsma dag_run.conf Var būt spēles mainītājs. Izmantojot DAG, nododot konfigurācijas vārdnīcu, jūs varat dinamiski ģenerēt uzdevumu secības. Tomēr tā ieviešanai strukturētā veidā ir nepieciešama dziļa izpratne par Airfow izpildes modeli.

Šajā rakstā mēs izpētīsim, kā izveidot dinamisku DAG, kur, izmantojot izpildlaiku, tiek noteiktas uzdevuma atkarības dag_run.confApvidū Ja esat cīnījies, lai to sasniegtu un neesat atradis skaidru risinājumu, neuztraucieties - jūs neesat viens! Sakārtosim to soli pa solim ar praktiskiem piemēriem. 🚀

Dinamisko dagu veidošana ar izpildlaika konfigurāciju gaisa plūsmā

Apache Airflow ir spēcīgs rīks sarežģītu darbplūsmu organizēšanai, taču tās patiesā izturība ir tā elastība. Iepriekš aprakstītie skripti parāda, kā izveidot a dinamiska DAG kur, izmantojot uzdevuma atkarības, tiek noteiktas izpildlaikā dag_run.confApvidū Tā vietā, lai kodētu apstrādājamos elementus, DAG tos dinamiski izgūst, kad tiek aktivizēts, ļaujot pielāgot vairāk darbplūsmas. Tas ir īpaši noderīgi reālās pasaules scenārijos, piemēram, mainīgu datu kopu apstrādes vai īpašu uzdevumu izpildē, pamatojoties uz ārējiem apstākļiem. Iedomājieties ETL cauruļvadu, kurā katru dienu mainās faili - šī pieeja padara automatizāciju daudz vieglāku. 🚀

Pirmais skripts izmanto Pythonoperators Dinamiski veikt uzdevumus un noteikt atkarības. Tas izvelk elementu sarakstu no dag_run.conf, Nodrošināt, ka uzdevumi tiek radīti tikai tad, kad nepieciešams. Katrs saraksta elements kļūst par unikālu uzdevumu, un atkarības tiek iestatītas secīgi. Otrā pieeja izmanto Tasklow API, kas vienkāršo DAG radīšanu ar tādiem dekoratoriem kā @dag un @TaskApvidū Šī metode padara DAG lasāmāku un uztur tīrāku izpildes loģiku. Šīs pieejas nodrošina, ka darbplūsmas var pielāgoties dažādām konfigurācijām, neprasot koda izmaiņas.

Piemēram, apsveriet scenāriju, kurā e-komercijas uzņēmums apstrādā pasūtījumus partijās. Dažās dienās var būt steidzamāki pasūtījumi nekā citām, kurām ir vajadzīgas dažādas uzdevumu secības. Statiskā DAG izmantošana nozīmētu koda modificēšanu katru reizi, kad mainās prioritātes. Izmantojot mūsu dinamisko DAG pieeju, ārējā sistēma var izraisīt DAG ar noteiktu uzdevumu secību, padarot procesu efektīvāku. Cits lietošanas gadījums ir datu zinātnē, kur modeļiem var būt nepieciešama pārkvalifikācija, pamatojoties uz ienākošajiem datu sadalījumiem. Dinamiski nododot nepieciešamās modeļa konfigurācijas, tiek izpildīti tikai nepieciešamie aprēķini, ietaupot laiku un resursus. 🎯

Rezumējot, šie skripti nodrošina pamatu DAG dinamiskai ģenerēšanai, pamatojoties uz izpildlaika ieejām. Piesaistot Gaisa plūsmas uzdevumu plūsmas API Vai arī tradicionālā pythonoperator pieeja, izstrādātāji var radīt elastīgas, modulāras un efektīvas darbplūsmas. Tas novērš nepieciešamību pēc manuālas iejaukšanās un ļauj nemanāmi integrēt citās automatizācijas sistēmās. Neatkarīgi no tā, vai apstrādā klientu pasūtījumus, pārvalda datu cauruļvadus vai organizē mākoņu darbplūsmas, dinamiskie DAG nodrošina gudrāku automatizāciju, kas pielāgota īpašām biznesa vajadzībām.

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

Viens

Dinamiskā uzdevumu sekvencēšanas uzlabošana ar nosacītu izpildi gaisa plūsmā

Viena spēcīga, bet bieži aizmirsta iezīme Apache gaisa plūsma ir nosacīta izpilde, kas var vēl vairāk uzlabot dinamiskās uzdevumu secības elastību. Vienlaikus izgūstot uzdevumu atkarības no dag_run.conf ir noderīga, reālās pasaules scenārijiem bieži ir jāveic tikai daži uzdevumi, pamatojoties uz īpašiem nosacījumiem. Piemēram, dažās datu kopās var būt nepieciešama priekšapstrāde pirms analīzes, bet citas var tikt tieši apstrādātas.

Nosacītu izpilde gaisa plūsmā var ieviest, izmantojot BranchPythonOperator, kas nosaka nākamo uzdevumu izpildīt, pamatojoties uz iepriekš noteiktu loģiku. Pieņemsim, ka mums ir dinamiska DAG, kas apstrādā failus, bet tikai failiem virs noteiktas izmērs ir nepieciešama validācija. Tā vietā, lai izpildītu visus uzdevumus secīgi, mēs varam dinamiski izlemt, kurus uzdevumus palaist, optimizējot izpildes laiku un samazinot resursu izmantošanu. Šī pieeja nodrošina, ka tiek iedarbināta tikai attiecīgā darbplūsma, padarot datu cauruļvadus efektīvākus. 🚀

Vēl viens veids, kā uzlabot dinamiskos DAG, ir, iekļaujot Viens (Starp komunikāciju ziņojumi). XCOM ļauj uzdevumiem apmainīties ar datiem, kas nozīmē, ka dinamiski izveidota uzdevumu secība var nodot informāciju starp soļiem. Piemēram, ETL cauruļvadā priekšapstrādes uzdevums var noteikt nepieciešamās pārvērtības un nodot šo informāciju turpmākajiem uzdevumiem. Šī metode ļauj patiesi balstītās darbplūsmas, kur izpildes plūsma pielāgojas, pamatojoties uz reālā laika ieejām, ievērojami palielinot automatizācijas iespējas.