Генерисање динамичких секвенци задатака у протоку ваздуха помоћу Цонфигурације Даг Рун

Откључавање снаге динамичких зависности од задатака у протоку ваздуха

Апацхе АирФлов је моћан алат за аутоматизацију радног тока, али руковање динамичким зависностима понекад се може осећати као решавање слагалице. Када дизајнирате усмерени ациклични графикон (ДАГ), радне секвенце за хардцодинг могу радити за једноставне случајеве употребе, али шта ако се структура мора одредити на време извођења? 🤔

Замислите да радите на цевоводу података где се задаци извршавају зависе од долазних података. На пример, обрада различитих комплета датотека на основу свакодневне конфигурације или извршавања променљивих трансформација на основу пословног правила. У таквим случајевима, статички даг неће је смањити - потребан вам је начин да динамично дефинишете зависности.

Ово је управо место где је ваздушни проток даг_рун.цонф може бити мењач игара. Проласком конфигурационог речника приликом покретања дага можете динамички створити секвенце задатака. Међутим, спровођење тога на структуриран начин захтева дубоко разумевање модела извршења ваздуха.

У овом чланку ћемо истражити како да изградимо динамичну даг где се одређују зависности од задатака на време извођења даг_рун.цонф. Ако се борите да то постигнете и нисте нашли јасан решење, не брините - нисте сами! Раздвојимо га корак по корак са практичним примерима. 🚀

Грађевина Динамиц Дагс са конфигурацијом рунтиме-а у протоку ваздуха

Апацхе АирФлов је моћан алат за оркестрирање сложених токова рада, али његова истинска снага лежи у својој флексибилности. Скрипте представљене раније показују како да креирају Динамиц Даг Тамо где се зависе од задатака у трајању даг_рун.цонф. Уместо да се у току листа елемената за обраду, даг их доноси динамички када се покреће, омогућавајући прилагодљивије радне токове рада. Ово је посебно корисно у сценаријима реалног света, као што су променљиви подаци о промјењивим подацима или извршавању специфичних задатака на основу спољних услова. Замислите етл цевовод где датотеке за обраду промјене дневно - овај приступ чини аутоматизацију много лакше. 🚀

Прва скрипта користи Питхоноператор да извршава задатке и динамички поставља зависности. Извади листу елемената из даг_рун.цонф, Осигуравање да се задаци креирају само по потреби. Сваки елемент на листи постаје јединствен задатак, а зависности су подешене узастопно. Други приступ утиче АПИ тобала задатака, који поједностављује креирање даго-а са декоратерима @даг и @таск. Ова метода чини ДАГ читљивијом и одржава чистију логику извршења. Ови приступи осигуравају да се радне токове могу прилагодити различитим конфигурацијама без потребе за промјенама кода.

На пример, размотрите сценариј у којем е-трговина компанија обрађује наређења у серијама. Неки дани могу имати хитније налоге од других, захтевајући различите секвенце задатака. Користећи статичку дагу значило би да се мења код сваки пут када се приоритети мењају. Помоћу нашег динамичког дагог приступа, спољни систем може покренути ДАГ са специфичним редоследом задатка, чинећи процес ефикаснији. Друга случај употребе је у науци о подацима, где ће модели можда требати преквалификацију на основу долазних дистрибуција података. Доношењем потребних конфигурација модела динамички, извршава се само потребно рачунање, штедећи време и ресурсе. 🎯

Укратко, ове скрипте пружају темеље за динамично генерисање дагова на основу уноса рунтиме. Искориштавањем АПИ тоба за ваздушне протокове Или традиционални приступ питоноператора, програмери могу да створе флексибилне, модуларне и ефикасне радне токове. Ово елиминише потребу за ручном интервенцијом и омогућава бешавну интеграцију са другим системима за аутоматизацију. Да ли обрада налога за купце, управљање цевоводима података или оркестрирајућим токовима облака, динамичне дагове омогућавају паметнију аутоматизацију прилагођену специфичним пословним потребама.

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

Повећавање динамичког секвизирања задатака са условним извршењем у протоку ваздуха

Један моћан, а опет често занемарен Апацхе Аирфлов је условно извршење, што може додатно побољшати флексибилност динамичког секвенцирања задатака. Док преузимају зависности од задатака даг_рун.цонф је користан, сценарији у стварном свету често захтевају извршавање само одређених задатака на основу одређених услова. На пример, неки датасетима могу захтевати прерађивање пре анализе, док се други могу директно обрадити.

Условно извршење у протоку ваздуха може се применити коришћењем BranchPythonOperator, што одређује следећи задатак да се изводи на основу унапред дефинисане логике. Претпоставимо да имамо динамичну дагу која процесуира датотеке, али само датотеке изнад одређене величине захтева валидацију. Уместо да извршете све задатке узастопно, можемо динамички одлучити које задатке који траје, оптимизујући време извршења и смањење употребе ресурса. Овај приступ осигурава да се активирају само релевантни токови рада, чинећи цевоводима података ефикаснијим. 🚀

Други начин за побољшање динамичних дагова је укључивањем XComs (Унакрсне комуникације). КСЦОМ-ови омогућавају задатке да размењују податке, што значи да динамички створена секвенца задатка може проћи информације између корака. На пример, у ЕТЛ цевоводу, задатак препровода може одредити тражене трансформације и пренети оне детаље на наредне задатке. Ова метода омогућава заиста радне токове погонских података, где се адапса протока извршења на основу уноса у реалном времену, значајно повећавајући могућности аутоматизације.