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 dag_run.conf 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 dag_run.conf. Č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. 🚀

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 Dinamični DAG kjer se odvisnosti od naloge določijo med izvajanjem dag_run.conf. 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 Pitonoperator Za izvajanje nalog in dinamično določiti odvisnosti. Izvleče seznam elementov iz dag_run.conf, 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 API -jev API, ki poenostavi ustvarjanje DAG z dekoraterji, kot so @dag in @Task. 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 API -jev Airflow AISFLOW 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 Apache Airflow je pogojna izvedba, ki lahko še izboljša prilagodljivost dinamičnega zaporedja nalog. Med pridobivanjem odvisnosti od nalog iz dag_run.conf 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 BranchPythonOperator, 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 XComs (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.