ഡാഗ് റൺ കോൺഫിഗറേഷൻ ഉപയോഗിച്ച് എയർലോയിൽ ഡൈനാമിക് ടാസ്ക് സീക്വൻസുകൾ സൃഷ്ടിക്കുന്നു

ചലനാത്മക ടാസ്ക് ഡിപൻഡൻസിയുടെ ശക്തി അൺലോക്കുചെയ്യുന്നു

അപരിചിതർ എയർഫോൺ ഒരു ശക്തമായ വർക്ക്ഫ്ലോ ഓട്ടോമേഷൻ ഉപകരണമാണ്, പക്ഷേ ചലനാത്മക ഡിപൻഡൻസികൾക്ക് കൈകാര്യം ചെയ്യാൻ ചിലപ്പോൾ ഒരു പസിൽ പരിഹരിക്കാനാകും. സംവിധാനം ചെയ്ത ACYCLic ഗ്രാഫ് (DAG) രൂപകൽപ്പന ചെയ്യുമ്പോൾ, ലളിതമായ ഉപയോഗ കേസുകൾക്കായി ഹാർഡ്കോഡിംഗ് ടാസ്ക് സീക്വൻസുകൾ പ്രവർത്തിച്ചേക്കാം, പക്ഷേ റൺടൈമിൽ ഘടന നിർണ്ണയിക്കണോ? പതനം

എക്സിക്യൂട്ട് ചെയ്യേണ്ട ജോലികൾ ഇൻകമിംഗ് ഡാറ്റയെ ആശ്രയിച്ചിരിക്കുന്നതിനെ ആശ്രയിച്ചിരുന്ന ഒരു ഡാറ്റ പൈപ്പ്ലൈനിൽ നിങ്ങൾ പ്രവർത്തിക്കുന്നുവെന്ന് സങ്കൽപ്പിക്കുക. ഉദാഹരണത്തിന്, ഒരു ബിസിനസ് നിയമത്തെ അടിസ്ഥാനമാക്കിയുള്ള ദൈനംദിന കോൺഫിഗറേഷൻ അല്ലെങ്കിൽ എക്സിക്യൂട്ട് പരിവർത്തനങ്ങൾ നടപ്പിലാക്കുന്ന വേരിയബിൾ പരിവർത്തനങ്ങൾ നടപ്പിലാക്കുന്ന അല്ലെങ്കിൽ വേരിയബിൾ പരിവർത്തനങ്ങൾ നടപ്പിലാക്കുക എന്നതിന് വ്യക്തമായ ഫയലുകളുടെ ഫയലുകൾ പ്രോസസ്സ് ചെയ്യുക. അത്തരം സാഹചര്യങ്ങളിൽ, ഒരു സ്റ്റാറ്റിക് ഡാഗ് അത് മുറിക്കില്ല - ഡിപറാനുകരണത്തെ മാനിക്കുന്ന നിർവചിക്കാൻ നിങ്ങൾക്ക് ഒരു വഴി ആവശ്യമാണ്.

ഇത് കൃത്യമായി എവിടെയാണ് വായുസഞ്ചാരം Dag_run.conf ഗെയിം മാറ്റുന്നയാൾ ആകാം. ഒരു ഡാഗിനെ പ്രേരിപ്പിക്കുമ്പോൾ ഒരു കോൺഫിഗറേഷൻ നിഘണ്ടു കൈമാറുന്നതിലൂടെ നിങ്ങൾക്ക് ടാസ്ക് സീക്വൻസുകൾ സൃഷ്ടിക്കാൻ കഴിയും. എന്നിരുന്നാലും, ഒരു ഘടനാപരമായ രീതിയിൽ ഇത് നടപ്പിലാക്കുന്നത് എയർഫോവ് എക്സിക്യൂഷൻ മോഡലിനെക്കുറിച്ച് ആഴത്തിലുള്ള ധാരണ ആവശ്യമാണ്.

ഈ ലേഖനത്തിൽ, റൺടൈമിൽ ടാസ്ക് ഡിപൻസികൾ നിർണ്ണയിക്കുന്ന ഒരു ഡൈനാമിക് ഡാഗ് എങ്ങനെ നിർമ്മിക്കാമെന്ന് ഞങ്ങൾ പര്യവേക്ഷണം ചെയ്യും Dag_run.conf. ഇത് നേടാൻ നിങ്ങൾ സമന്വയിപ്പിക്കുകയും വ്യക്തമായ പരിഹാരം കണ്ടെത്തുകയും ചെയ്തില്ലെങ്കിൽ, വിഷമിക്കേണ്ട - നിങ്ങൾ ഒറ്റയ്ക്കല്ല! പ്രായോഗിക ഉദാഹരണങ്ങളിലൂടെ ഘട്ടം ഘട്ടമായി ഇത് തകർക്കാം. പതനം

എയർലോയിൽ റൺടൈം കോൺഫിഗറേഷൻ ഉപയോഗിച്ച് ഡൈനാമിക് ഡാഗുകൾ നിർമ്മിക്കുക

കോംപ്ലന്റ് വർക്ക്ഫ്ലോറുകളുടെ ശക്തമായ ഒരു ഉപകരണമാണ് അപ്പാച്ചെ എയർഫോൺ, പക്ഷേ അതിന്റെ യഥാർത്ഥ ശക്തി അതിന്റെ വഴക്കത്തിലാണ്. മുമ്പ് അവതരിപ്പിച്ച സ്ക്രിപ്റ്റുകൾ ഒരു സൃഷ്ടിക്കാമെന്ന് പ്രദർശിപ്പിക്കുന്നു ഡൈനാമിക് ഡാഗ് റൺടൈമിൽ ടാസ്ക് ഡിപൻസികൾ നിർണ്ണയിക്കുന്ന ഇടം Dag_run.conf. പ്രോസസ്സ് ചെയ്യുന്നതിനുള്ള മൂലകങ്ങളുടെ പട്ടിക ഹാർഡ്കോഡുചെയ്തതിന് പകരം, കൂടുതൽ പൊരുത്തപ്പെടുന്ന വർക്ക്ഫ്ലോറുകൾ അനുവദിക്കുമ്പോൾ ഡാഗ് അവ ചലനാത്മകമായി ചൂടാക്കുന്നു. ഇത് പ്രോസസ്സിംഗ് വേരിയബിൾ ഡാറ്റാസെറ്റുകൾ അല്ലെങ്കിൽ ബാഹ്യ സാഹചര്യങ്ങളെ അടിസ്ഥാനമാക്കി പ്രോസസിംഗ് വേരിയബിൾ ഡാറ്റാസെറ്റുകൾ പോലുള്ള യഥാർത്ഥ ലോക സാഹചര്യങ്ങളിൽ ഇത് പ്രത്യേകിച്ചും ഉപയോഗപ്രദമാണ്. പ്രതിദിനം പ്രോസസ്സ് ചെയ്യുന്നതിനുള്ള ഫയലുകൾ ദൈനംദിന മാറ്റം വരുത്തുന്ന ഒരു എറ്റ് പൈപ്പ്ലൈൻ സങ്കൽപ്പിക്കുക - ഈ സമീപനം ഓട്ടോമേഷൻ വളരെ എളുപ്പമാക്കുന്നു. പതനം

ആദ്യത്തെ സ്ക്രിപ്റ്റ് ഉപയോഗിക്കുന്നു പൈടോൺപെററ്റർ ചുമതലകൾ നടപ്പിലാക്കുന്നതിനും ഡിപൻഡൈസികൾ ചലനാത്മകമായി സജ്ജീകരിക്കുന്നതിനും. ഇത് മൂലകങ്ങളുടെ പട്ടികയിൽ നിന്ന് പുറത്തെടുക്കുന്നു Dag_run.conf, ആവശ്യമുള്ളപ്പോൾ മാത്രം ജോലി സൃഷ്ടിക്കപ്പെടുന്നതായി ഉറപ്പാക്കുന്നു. പട്ടികയിലെ ഓരോ ഘടകവും ഒരു അദ്വിതീയ ജോലിയായി മാറുന്നു, ഡിപൻഡൻസികൾ തുടർച്ചയായി സജ്ജമാക്കി. രണ്ടാമത്തെ സമീപനം പ്രയോജനപ്പെടുത്തുന്നു ടാസ്ക്ഫ്ലോ API, ഇത് ഡാഗ് സൃഷ്ടിയെ അലങ്കാരക്കാരെ ലളിതമാക്കുന്നു @DAG കൂടെ ast ടാസ്ക്. ഈ രീതി ഡാഗ് കൂടുതൽ വായിക്കാൻ കഴിയുന്നതും ക്ലീനർ എക്സിക്യൂഷൻ യുക്തി നിലനിർത്തുകയും ചെയ്യുന്നു. കോഡ് മാറ്റങ്ങൾ ആവശ്യമില്ലാതെ വർക്ക് ഫ്ലോറുകൾ വ്യത്യസ്ത കോൺഫിഗറേഷനുകളുമായി പൊരുത്തപ്പെടുത്താൻ ഈ സമീപനങ്ങൾ ഉറപ്പാക്കുന്നു.

ഉദാഹരണത്തിന്, ഒരു ഇ-കൊമേഴ്സ് കമ്പനി ബാച്ചുകളായി ഓർഡറുകൾ പ്രോസസ്സ് ചെയ്യുന്ന ഒരു സാഹചര്യം പരിഗണിക്കുക. വ്യത്യസ്ത ടാസ്ക് സീക്വൻസുകൾ ആവശ്യമുള്ളതിനാൽ ചില ദിവസങ്ങൾക്ക് മറ്റുള്ളവയേക്കാൾ അടിയന്തിര ഓർഡറുകൾ ഉണ്ടായിരിക്കാം. ഒരു സ്റ്റാറ്റിക് ഡാഗ് ഉപയോഗിച്ച് ഓരോ തവണയും മുൻഗണനകൾ മാറുന്നത് കോഡ് പരിഷ്കരിക്കുക എന്നാണ്. ഞങ്ങളുടെ ഡൈനാമിക് ഡാഗ് സമീപനത്തോടെ, ഒരു ബാഹ്യ സിസ്റ്റത്തിന് ഒരു നിർദ്ദിഷ്ട ടാസ്ക് ശ്രേണി ഉപയോഗിച്ച് ഡാഗ് പ്രവർത്തനക്ഷമമാക്കാം, പ്രക്രിയ കൂടുതൽ കാര്യക്ഷമമാക്കുന്നു. മറ്റൊരു ഉപയോഗ കേസ് ഡാറ്റ സയൻസിലാണ്, അവിടെ ഇൻകമിംഗ് ഡാറ്റ വിതരണങ്ങളെ അടിസ്ഥാനമാക്കി മോഡലുകൾക്ക് വീണ്ടും റിട്ടേണിംഗ് ആവശ്യമായി വന്നേക്കാം. ആവശ്യമായ മോഡൽ കോൺഫിഗറേഷനുകൾ ചലനാത്മകമായി കടന്നുപോകുന്നതിലൂടെ, ആവശ്യമായ കണക്കുകൂട്ടലുകൾ മാത്രമേ നടപ്പിലാക്കൂ, സമയം, വിഭവങ്ങൾ എന്നിവ പരിഹരിക്കുകയാണ്. പതനം

ചുരുക്കത്തിൽ, ഈ സ്ക്രിപ്റ്റുകൾ റൺടൈം ഇൻപുട്ടിനെ അടിസ്ഥാനമാക്കിയുള്ള ഡാഗുകൾ ചലനാത്മകമായി ഉൽപാദിപ്പിക്കുന്ന ഒരു അടിത്തറ നൽകുന്നു. ഒഴിവാക്കുന്നതിലൂടെ വായുസഞ്ചാരത്തിന്റെ ടാസ്ക്ഫ്ലോ API അല്ലെങ്കിൽ പരമ്പരാഗത പൈത്തൊൺപോർപറേറ്റർ സമീപനം, ഡവലപ്പർമാർക്ക് വഴക്കമുള്ള, മോഡുലാർ, കാര്യക്ഷമമായ വർക്ക്ഫ്ലകൾ സൃഷ്ടിക്കാൻ കഴിയും. ഇത് സ്വമേധയാലുള്ള ഇടപെടലിന്റെ ആവശ്യകത ഇല്ലാതാക്കുകയും മറ്റ് ഓട്ടോമേഷൻ സിസ്റ്റങ്ങളുമായി തടസ്സമില്ലാത്ത സംയോജനത്തിന് അനുവദിക്കുകയും ചെയ്യുന്നു. ഉപഭോക്തൃ ഓർഡറുകൾ പ്രോത്സാഹിപ്പിക്കുക, ഡാറ്റ പൈപ്പ്ലൈനുകൾ കൈകാര്യം ചെയ്യുക, അല്ലെങ്കിൽ ഓർക്കസ്ട്രേറ്റിംഗ് ക്ലൗഡ് വർക്ക്ഫ്ലോകൾ, ഡൈനാമിക് ഡാഗുകൾ നിർദ്ദിഷ്ട ബിസിനസ് ആവശ്യങ്ങൾക്ക് അനുയോജ്യമായ മികച്ച ഓട്ടോമേഷൻ പ്രാപ്തമാക്കുന്നു.

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

വായുസഞ്ചാരത്ത് സോപാധിക വധശിക്ഷയോടെ ഡൈനാമിക് ടാസ്ക് വർദ്ധിപ്പിക്കുന്നു

ഒരു ശക്തമായ സവിശേഷത പലപ്പോഴും അവഗണിക്കപ്പെട്ടു അപ്പാച്ചെ വായുസഞ്ചാരം സോപാധികമായ വധശിക്ഷയാണ്, ഇത് ഡൈനാമിക് ടാസ്ക് സീക്വൻസിയുടെ വഴക്കം മെച്ചപ്പെടുത്താൻ കഴിയും. ടാസ്ക് ഡിപൻഡൻസികൾ വീണ്ടെടുക്കുമ്പോൾ Dag_run.conf ഉപയോഗപ്രദമാണ്, യഥാർത്ഥ സാഹചര്യങ്ങളിൽ പലപ്പോഴും നിർദ്ദിഷ്ട സാഹചര്യങ്ങളെ അടിസ്ഥാനമാക്കി ചില ജോലികൾ മാത്രമേ എക്സിക്യൂട്ട് ചെയ്യേണ്ടൂ. ഉദാഹരണത്തിന്, ചില ഡാറ്റാസെറ്റുകൾ വിശകലനത്തിന് മുമ്പ് പ്രീപ്രൊസസിംഗ് ആവശ്യമായി വന്നേക്കാം, മറ്റുള്ളവ നേരിട്ട് പ്രോസസ്സ് ചെയ്യാൻ കഴിയും.

വായുസഞ്ചാരത്തിലെ സോപാധികമായ വധശിക്ഷ നടപ്പാക്കാം BranchPythonOperator, മുൻനിശ്ചയിച്ച യുക്തിയെ അടിസ്ഥാനമാക്കി നിർവ്വഹിക്കാനുള്ള അടുത്ത ടാസ്ക് നിർണ്ണയിക്കുന്ന ഇത്. ഫയലുകൾ പ്രോസസ്സ് ചെയ്യുന്ന ഒരു ഡൈനാമിക് ഡാഗ് ഉണ്ടെന്ന് കരുതുക, ഒരു നിശ്ചിത വലുപ്പത്തിന് മുകളിലുള്ള ഫയലുകൾക്ക് മാത്രമാണ് മൂല്യനിർണ്ണയം ആവശ്യമാണ്. എല്ലാ ജോലികളും തുടർച്ചയായി നടപ്പിലാക്കുന്നതിനുപകരം, ഏത് ടാസ്ക്കുകൾ ആസൂത്രണം ചെയ്യാനും നിർവ്വഹണ സമയത്തെ ഒപ്റ്റിമൈമാക്കാനും റിസോഴ്സ് ഉപയോഗം കുറയ്ക്കാനും ഞങ്ങൾക്ക് കഴിയും. ഡാറ്റ പൈപ്പ്ലൈനുകൾ കൂടുതൽ കാര്യക്ഷമമാക്കുന്നതിലൂടെ പ്രസക്തമായ വർക്ക്ഫ്ലോകൾ മാത്രമേ പ്രസക്തമായ വർക്ക് ലൈഫ് പരീക്ഷിക്കുകയും ഈ സമീപനം ഉറപ്പാക്കുന്നു. പതനം

ഡൈനാമിക് ഡാഗുകൾ വർദ്ധിപ്പിക്കുന്നതിനുള്ള മറ്റൊരു മാർഗം സംയോജിപ്പിക്കുന്നതിലൂടെയാണ് XComs (ക്രോസ്-ആശയവിനിമയ സന്ദേശങ്ങൾ). എക്സ്കോമുകൾ ഡാറ്റ കൈമാറാൻ ടാസ്ക്കുകൾ അനുവദിക്കുക, അതായത് ചലനാത്മകമായി സൃഷ്ടിച്ച ടാസ്ക് സീക്വൻസ് ഘട്ടങ്ങൾക്കിടയിൽ വിവരങ്ങൾ കൈമാറാൻ കഴിയും. ഉദാഹരണത്തിന്, ഒരു എറ്റ് പൈപ്പ്ലൈനിൽ, ഒരു പ്രീപ്രൊസസിംഗ് ടാസ്ക് ആവശ്യമായ പരിവർത്തനങ്ങൾ നിർണ്ണയിക്കുകയും അവ തുടർന്നുള്ള ചുമതലകൾക്ക് കൈമാറുകയും ചെയ്യും. ഈ രീതി യഥാർത്ഥത്തിൽ ഡാറ്റ നയിക്കുന്ന വർക്ക്ഫ്ലകളെ പ്രാപ്തമാക്കുന്നു, അവിടെ വധശിക്ഷാ പ്രവാഹം, തത്സമയ ഇൻപുട്ടുകൾ അടിസ്ഥാനമാക്കി, ഓട്ടോമേഷൻ കഴിവുകൾ ഗണ്യമായി വർദ്ധിപ്പിക്കുക.