Dynaamisten tehtäväsekvenssien luominen ilmavirrassa DAG RUN -konfiguraatiolla

Dynaamisten tehtäväriippuvuuksien voiman avaaminen ilmavirrassa

Apache AirFlow on tehokas työnkulun automaatiotyökalu, mutta dynaamisten riippuvuuksien käsittely voi joskus tuntua ratkaistavan palapelin. Suunnitellessasi suunnattua asyklistä kuvaajaa (DAG), kovakoodaustehtäväsekvenssit saattavat toimia yksinkertaisissa käyttötapauksissa, mutta entä jos rakenne on määritettävä suorituksen aikana? 🤔

Kuvittele, että työskentelet tietoputken parissa, jossa suoritettavat tehtävät riippuvat tulevista tiedoista. Esimerkiksi erilaisten tiedostojoukkojen käsittely päivittäisen kokoonpanon perusteella tai muuttuvien muunnosten suorittaminen liiketoimintasäännän perusteella. Tällaisissa tapauksissa staattinen DAG ei leikkaa sitä - tarvitset tapaa määritellä riippuvuudet dynaamisesti.

Tässä on juuri ilmavirta dag_run.conf Voi olla pelinvaihtaja. Siirtämällä kokoonpanosanakirja DAG: n käynnistäessä voit luoda dynaamisesti tehtäväsekvenssejä. Tämän toteuttaminen rakenteellisella tavalla vaatii kuitenkin syvän ymmärryksen ilmavirran suoritusmallista.

Tässä artikkelissa tutkimme kuinka rakentaa dynaaminen DAG, jossa tehtäväriippuvuudet määritetään ajon aikana dag_run.conf. Jos olet pyrkinyt saavuttamaan tämän etkä ole löytänyt selkeää ratkaisua, älä huoli - et ole yksin! Hajotellaan se askel askeleelta käytännön esimerkkien kanssa. 🚀

Dynaamisten DAG: ien rakentaminen ajonaikaisella kokoonpanolla ilmavirrassa

Apache AirFlow on tehokas työkalu monimutkaisten työnkulkujen orkestrointiin, mutta sen todellinen vahvuus on sen joustavuudessa. Aikaisemmin esitetyt skriptit osoittavat kuinka luoda a dynaaminen DAG missä tehtäväriippuvuudet määritetään ajon aikana käyttämällä dag_run.conf. Sen sijaan, että koodataan koodaus elementtiluetteloon, DAG hakee ne dynaamisesti laukaisun yhteydessä, mikä mahdollistaa mukautuvammat työnkulut. Tämä on erityisen hyödyllistä reaalimaailman skenaarioissa, kuten muuttuvien tietojoukkojen käsittely tai ulkoisten olosuhteiden perusteella tiettyjen tehtävien suorittaminen. Kuvittele ETL -putkilinja, jossa tiedostot vaihtavat päivittäin - tämä lähestymistapa helpottaa automaatiota. 🚀

Ensimmäinen käsikirjoitus käyttää Pythonoperator Tehtävien suorittaminen ja riippuvuuksien asettaminen dynaamisesti. Se purkaa elementtiluettelon dag_run.conf, varmistaa, että tehtävät luodaan vain tarvittaessa. Jokaisesta luettelon elementistä tulee ainutlaatuinen tehtävä, ja riippuvuudet asetetaan peräkkäin. Toinen lähestymistapa hyödyntää Taskflow API, mikä yksinkertaistaa DAG: n luomista sisustajien kanssa @dag ja @tehtävä. Tämä menetelmä tekee DAG: sta luettavissa ja ylläpitää puhdistusaineiden suorituslogiikkaa. Nämä lähestymistavat varmistavat, että työnkulkut voivat sopeutua erilaisiin kokoonpanoihin vaatimatta koodimuutoksia.

Harkitse esimerkiksi skenaariota, jossa verkkokauppayritys käsittelee tilauksia erissä. Joillakin päivinä voi olla kiireellisempiä tilauksia kuin toisilla, jotka vaativat erilaisia ​​tehtäväsekvenssejä. Staattisen DAG: n käyttäminen tarkoittaisi koodin muuttamista joka kerta, kun prioriteetit muuttuvat. Dynaamisella DAG -lähestymistavalla ulkoinen järjestelmä voi laukaista DAG: n tietyllä tehtäväsekvenssillä, mikä tekee prosessista tehokkaamman. Toinen käyttötapa on tietotekniikassa, jossa malleja saattavat tarvita uudelleenkokoonpanon saapuvien tietojakaumien perusteella. Ohittamalla vaadittavat malliskonfiguraatiot dynaamisesti, suoritetaan vain tarvittavat laskelmat säästäen aikaa ja resursseja. 🎯

Yhteenvetona voidaan todeta, että nämä skriptit tarjoavat perustan dynaamisesti DAG: ien tuottamiselle, joka perustuu ajoajan tuloihin. Hyödyntämällä Ilmavirran Taskflow -sovellusliittymä Tai perinteinen Pythonoperator -lähestymistapa, kehittäjät voivat luoda joustavia, modulaarisia ja tehokkaita työnkulkuja. Tämä eliminoi manuaalisen intervention tarpeen ja mahdollistaa saumattoman integroinnin muihin automaatiojärjestelmiin. Dynaamiset DAG: t mahdollistavat asiakkaiden tilausten käsittely, dataputkien hallinta tai pilvien työnkulkujen järjestäminen, mikä mahdollistaa tietyille liiketoiminnan tarpeille räätälöidyn älykkäämmän automaation.

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

Dynaamisen tehtäväsekvensoinnin parantaminen ehdollisella suorituksella ilmavirrassa

Yksi voimakas, mutta usein huomiotta jätetty ominaisuus Apache -ilmavirta on ehdollinen suoritus, joka voi edelleen parantaa dynaamisen tehtäväsekvensoinnin joustavuutta. Haettaessa tehtäväriippuvuuksia dag_run.conf on hyödyllistä, reaalimaailman skenaariot vaativat usein tietyn tiettyjen olosuhteiden perusteella vain tiettyjen tehtävien suorittamista. Esimerkiksi jotkut tietojoukot saattavat vaatia esikäsittelyä ennen analyysiä, kun taas toiset voidaan käsitellä suoraan.

Ehdollinen suoritus ilmavirrassa voidaan toteuttaa käyttämällä 0 -, joka määrittää seuraavan suoritettavan tehtävän ennalta määritetyn logiikan perusteella. Oletetaan, että meillä on dynaaminen DAG, joka käsittelee tiedostoja, mutta vain tietyn koon yläpuolella olevat tiedostot vaativat validointia. Sen sijaan, että suoritettaisiin kaikki tehtävät peräkkäin, voimme dynaamisesti päättää, mitkä tehtävät suoritetaan, optimoimalla suoritusaika ja vähentämällä resurssien käyttöä. Tämä lähestymistapa varmistaa, että vain merkitykselliset työnkulut laukaistaan, mikä tekee tietoputkista tehokkaamman. 🚀

Toinen tapa parantaa dynaamisia DAG: itä on sisällyttämällä XComs (Ristiviestit). XCOM: t sallivat tehtävien vaihtaa tietoja, mikä tarkoittaa, että dynaamisesti luotu tehtäväsekvenssi voi välittää tietoja vaiheiden välillä. Esimerkiksi ETL -putkilinjassa esikäsittelytehtävä voi määrittää vaadittavat muunnokset ja siirtää nämä yksityiskohdat seuraaviin tehtäviin. Tämä menetelmä mahdollistaa todella tietopohjaiset työnkulut, joissa suoritusvirta sovittuu reaaliaikaisten tulojen perusteella lisäämällä automaatiokykyä merkittävästi.