Ustvarjanje dinamičnih zaporedja nalog v pretoku zraka z uporabo konfiguracije DAG Run

Odklepanje moči dinamičnih odvisnosti od nalog v pretoku zraka

Apache Airflow je močno orodje za avtomatizacijo delovnega toka, vendar se lahko obvladovanje dinamičnih odvisnosti včasih zdi, kot da bi rešili sestavljanko. Pri oblikovanju usmerjenega acikličnega grafa (DAG) lahko trde kodiranja zaporedja nalog delujejo za preproste primere uporabe, kaj pa, če je treba strukturo določiti med izvajanjem? 🤔

Predstavljajte si, da delate na podatkovnem plinovodu, kjer so naloge, ki jih je treba opraviti, odvisno od dohodnih podatkov. Na primer, obdelava različnih nizov datotek, ki temelji na dnevni konfiguraciji ali izvajanju spremenljivih preobrazb na podlagi poslovnega pravila. V takih primerih ga statični DAG ne bo zmanjšal - potrebujete način, da dinamično določite odvisnosti.

Prav tam je zračni pretok Lahko je menjalnik iger. S prehodom konfiguracijskega slovarja pri sprožitvi DAG lahko dinamično ustvarite zaporedja opravil. Vendar pa je za to, da to izvajate na strukturiran način, globoko razumevanje modela izvajanja Airflow.

V tem članku bomo raziskali, kako sestaviti dinamičen DAG, kjer se odvisnosti nalog določajo med izvajanjem . Če ste se trudili, da bi to dosegli in niste našli jasne rešitve, ne skrbite - niste sami! Razčlenimo korak za korakom s praktičnimi primeri. 🚀

Ukaz	Primer uporabe
dag_run.conf	Omogoča pridobivanje dinamičnih konfiguracijskih vrednosti pri sprožitvi teka DAG. Bistvenega pomena za prehod parametrov izvajanja.
PythonOperator	Določi nalogo v pretoku zraka, ki izvaja funkcijo Python, kar omogoča fleksibilno logiko izvajanja znotraj DAG.
set_upstream()	Izrecno definira odvisnost med nalogami in tako zagotovi, da se ena naloga izvaja šele za tem, da se konča.
@dag	Dekorater, ki ga je ponudil API TASKTFOW, da definira DAGS na bolj pitoničen in strukturiran način.
@task	Omogoča določitev nalog v pretoku zraka z uporabo API -jev nalog, poenostavi ustvarjanje nalog in posredovanje podatkov.
override(task_id=...)	Uporablja se za dinamično spreminjanje ID -ja naloge, ko zaupne več nalog iz ene same funkcije.
extract_elements(dag_run=None)	Funkcija, ki izvleče vrednosti iz slovarja dag_run.conf, da dinamično konfigurira izvajanje opravil.
schedule_interval=None	Zagotavlja, da se DAG izvede le, ko se ročno sproži, namesto da bi tekel po fiksnem urniku.
op_args=[element]	Dinamične argumente prenese na nalogo pitonoperatorja, ki omogoča različne izvedbe na na primer opravilo.
catchup=False	Preprečuje, da bi zračni tok izvajal vse zgrešene usmrtitve DAG, ko se začne po pavzi, uporabna za konfiguracije v realnem času.

Gradnja dinamičnih dags s konfiguracijo izvajanja v pretoku zraka

Apache Airflow je močno orodje za orkestriranje zapletenih delovnih tokov, vendar je njegova resnična moč v njegovi prilagodljivosti. Prej predstavljeni skripti prikazujejo, kako ustvariti kjer se odvisnosti od naloge določijo med izvajanjem . Namesto da bi trdo kodirali seznam elementov, ki jih je treba obdelati, jih DAG, ko se sproži, dinamično pridobiva, kar omogoča bolj prilagodljive delovne tokove. To je še posebej koristno v scenarijih v resničnem svetu, kot so obdelava naborov spremenljivk ali izvajanje posebnih nalog na podlagi zunanjih pogojev. Predstavljajte si cevovod ETL, kjer se datoteke za obdelavo vsakodnevno spreminjajo - ta pristop olajša avtomatizacijo. 🚀

Prvi skript uporablja Za izvajanje nalog in dinamično določiti odvisnosti. Izvleče seznam elementov iz , zagotavljanje, da se naloge ustvarjajo samo, kadar je to potrebno. Vsak element na seznamu postane edinstvena naloga, odvisnosti pa so nastavljene zaporedno. Drugi pristop izkorišča , ki poenostavi ustvarjanje DAG z dekoraterji, kot so @dag in . Ta metoda naredi DAG bolj berljiv in vzdržuje čistejšo logiko izvajanja. Ti pristopi zagotavljajo, da se lahko delovni tokovi prilagodijo različnim konfiguracijam, ne da bi potrebovali spremembe kode.

Na primer, razmislite o scenariju, kjer podjetje za e-trgovino obdeluje naročila v serijah. Nekateri dnevi imajo lahko bolj nujna naročila kot drugi, ki zahtevajo različne zaporedje nalog. Uporaba statičnega DAG bi pomenila spreminjanje kode vsakič, ko se prednostne naloge spremenijo. Z našim dinamičnim pristopom DAG lahko zunanji sistem sproži DAG z določenim zaporedjem opravil, zaradi česar je postopek učinkovitejši. Drug primer uporabe je v podatkovni znanosti, kjer bodo modeli morda potrebovali prekvalifikacijo na podlagi dohodnih porazdelitev podatkov. Z dinamično prenašanjem potrebnih konfiguracij modela se izvajajo le potrebni izračuni, s čimer se prihrani čas in vire. 🎯

Če povzamemo, ti skripti predstavljajo temelje za dinamično ustvarjanje DAG, ki temeljijo na vhodih izvajanja. Z uporabo Ali tradicionalni pristop pitonoperatorja lahko razvijalci ustvarijo prilagodljive, modularne in učinkovite delovne tokove. To odpravlja potrebo po ročnem intervenciji in omogoča brezhibno integracijo z drugimi sistemi za avtomatizacijo. Ne glede na to, ali obdelate naročila strank, upravljanje podatkovnih cevovodov ali orkestrirajo v oblaku, dinamični DAG omogočajo pametnejšo avtomatizacijo, prilagojeno določenim poslovnim potrebam.

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

Izboljšanje dinamičnega zaporedja nalog s pogojno izvedbo v pretoku zraka

Ena močna, a pogosto spregledana funkcija v je pogojna izvedba, ki lahko še izboljša prilagodljivost dinamičnega zaporedja nalog. Med pridobivanjem odvisnosti od nalog iz je koristno, scenariji v resničnem svetu pogosto zahtevajo izvajanje samo določenih nalog na podlagi posebnih pogojev. Na primer, nekatere nabore podatkov bodo morda zahtevale predhodno obdelavo pred analizo, druge pa je mogoče obdelati neposredno.

Pogojno izvajanje v zračnem toku je mogoče izvesti s pomočjo , ki določa naslednjo nalogo, ki jo izvajate na podlagi vnaprej določene logike. Recimo, da imamo dinamičen DAG, ki obdeluje datoteke, vendar le datoteke nad določeno velikostjo zahtevajo preverjanje. Namesto da bi vse naloge izvajali zaporedno, se lahko dinamično odločimo, katere naloge izvajati, optimiziramo čas izvedbe in zmanjšamo porabo virov. Ta pristop zagotavlja, da se sprožijo samo ustrezni delovni tokovi, kar omogoča učinkovitejše podatke. 🚀

Drug način za izboljšanje dinamičnih DAG je z vključitvijo (Med komunikacijska sporočila). XCOM omogočajo naloge za izmenjavo podatkov, kar pomeni, da lahko dinamično ustvarjeno zaporedje opravil prenese informacije med koraki. Na primer, v cevovodu ETL lahko naloga predhodne obdelave določi zahtevane preobrazbe in te podrobnosti prenese na naslednje naloge. Ta metoda omogoča resnično delovne tokove, ki temeljijo na podatkih, kjer se pretok izvedbe prilagodi na podlagi vhodov v realnem času, kar je znatno povečalo zmogljivosti avtomatizacije.

1. Kaj je se uporablja za?
2. Omogoča prehod konfiguracijskih parametrov med izvajanjem, ko sproži DAG, zaradi česar so delovni tokovi bolj prilagodljivi.
3. Kako lahko dinamično ustvarjam naloge v zračnem toku?
4. Lahko uporabite zanko, da sprožite več primerkov a ali uporabite Dekorater v API -ju Taskflow.
5. Kaj je prednost uporabe ?
6. Omogoča pogojno izvedbo, kar omogoča DAG -jem, da sledijo različnim potm, ki temeljijo na vnaprej določeni logiki, kar izboljšuje učinkovitost.
7. Kako Izboljšajte dinamične DAGS?
8. XCOM omogočajo naloge za izmenjavo podatkov in zagotavljajo, da nadaljnje naloge prejemajo ustrezne informacije iz prejšnjih korakov.
9. Ali lahko dinamično nastavim odvisnosti?
10. Da, lahko uporabite in metode za dinamično opredelitev odvisnosti znotraj DAG.

Izvajanje V zračnem toku bistveno poveča avtomatizacijo delovnega toka, zaradi česar je prilagodljiv spreminjajočim se zahtevam. Z izkoriščanjem konfiguracij izvajanja se lahko razvijalci izognejo statičnim definicijam DAG in namesto tega ustvarijo prilagodljive cevovode, ki temeljijo na podatkih. Ta pristop je še posebej dragocen v okoljih, kjer je treba naloge določiti na podlagi vnosa v realnem času, kot sta finančno poročanje ali usposabljanje modela strojnega učenja. 🎯

Z integracijo , pogojna izvedba in upravljanje odvisnosti, ekipe lahko ustvarijo razširljive in učinkovite delovne tokove. Ne glede na to, ali obdelava transakcij e-trgovine, upravljanje transformacij podatkov v oblaku ali orkestriranje zapletenih serijskih opravil, dinamične zmogljivosti DAG Airflow zagotavljajo optimizirano in avtomatizirano rešitev. Vlaganje v te tehnike podjetjem omogoča racionalizacijo operacij, hkrati pa zmanjšuje ročno posredovanje.