DAG ரன் உள்ளமைவைப்

Alice Dupont

வியாழன், 13 பிப்ரவரி, 2025 ’அன்று’ பிற்பகல் 8:24:08

காற்றோட்டத்தில் டைனமிக் பணி சார்புகளின் சக்தியைத் திறத்தல்

அப்பாச்சி காற்றோட்டம் ஒரு சக்திவாய்ந்த பணிப்பாய்வு ஆட்டோமேஷன் கருவியாகும், ஆனால் மாறும் சார்புகளை கையாள்வது சில நேரங்களில் ஒரு புதிரைத் தீர்ப்பது போல் உணரலாம். இயக்கிய அசைக்ளிக் வரைபடத்தை (டிஏஜி) வடிவமைக்கும்போது, ஹார்ட்கோடிங் பணி வரிசைமுறைகள் எளிய பயன்பாட்டு நிகழ்வுகளுக்கு வேலை செய்யக்கூடும், ஆனால் இயக்க நேரத்தில் கட்டமைப்பை தீர்மானிக்க வேண்டும் என்றால் என்ன செய்வது? .

உள்வரும் தரவைப் பொறுத்து செயல்படுத்தப்பட வேண்டிய பணிகள் ஒரு தரவுக் குழாயில் நீங்கள் வேலை செய்கிறீர்கள் என்று கற்பனை செய்து பாருங்கள். எடுத்துக்காட்டாக, தினசரி உள்ளமைவின் அடிப்படையில் வெவ்வேறு கோப்புகளை செயலாக்குவது அல்லது வணிக விதியின் அடிப்படையில் மாறி மாற்றங்களை செயல்படுத்துதல். இதுபோன்ற சந்தர்ப்பங்களில், ஒரு நிலையான DAG அதைக் குறைக்காது the சார்புகளை மாறும் வகையில் வரையறுக்க உங்களுக்கு ஒரு வழி தேவை.

இது துல்லியமாக காற்றோட்டமாகும் Dag_run.conf விளையாட்டு மாற்றியாக இருக்கலாம். ஒரு DAG ஐத் தூண்டும் போது ஒரு உள்ளமைவு அகராதியை அனுப்புவதன் மூலம், நீங்கள் பணி காட்சிகளை மாறும் வகையில் உருவாக்கலாம். இருப்பினும், இதை ஒரு கட்டமைக்கப்பட்ட வழியில் செயல்படுத்துவதற்கு காற்றோட்டத்தின் மரணதண்டனை மாதிரியைப் பற்றிய ஆழமான புரிதல் தேவைப்படுகிறது.

இந்த கட்டுரையில், ஒரு டைனமிக் டிஏஜியை எவ்வாறு உருவாக்குவது என்பதை ஆராய்வோம், அங்கு பணி சார்புநிலைகள் இயக்க நேரத்தில் தீர்மானிக்கப்படுகின்றன dag_run.conf. இதை அடைய நீங்கள் சிரமப்படுகிறீர்கள் மற்றும் தெளிவான தீர்வைக் காணவில்லை என்றால், கவலைப்பட வேண்டாம் - நீங்கள் தனியாக இல்லை! நடைமுறை எடுத்துக்காட்டுகளுடன் படிப்படியாக அதை உடைப்போம். .

கட்டளை	பயன்பாட்டின் எடுத்துக்காட்டு
dag_run.conf	ஒரு DAG ஓட்டத்தைத் தூண்டும் போது டைனமிக் உள்ளமைவு மதிப்புகளை மீட்டெடுக்க அனுமதிக்கிறது. இயக்க நேர அளவுருக்களைக் கடக்க அவசியம்.
PythonOperator	பைதான் செயல்பாட்டை செயல்படுத்தும் காற்றோட்டத்தில் ஒரு பணியை வரையறுக்கிறது, இது ஒரு DAG க்குள் நெகிழ்வான மரணதண்டனை தர்க்கத்தை அனுமதிக்கிறது.
set_upstream()	பணிகளுக்கு இடையிலான சார்புநிலையை வெளிப்படையாக வரையறுக்கிறது, ஒரு பணி மற்றொன்றுக்கு மட்டுமே முடிந்துவிட்டது என்பதை உறுதி செய்கிறது.
@dag	DAG களை மிகவும் பைத்தானிக் மற்றும் கட்டமைக்கப்பட்ட வழியில் வரையறுக்க டாஸ்க்ஃப்ளோ ஏபிஐ வழங்கிய ஒரு அலங்காரக்காரர்.
@task	டாஸ்க்ஃப்ளோ ஏபிஐ பயன்படுத்தி காற்றோட்டத்தில் பணிகளை வரையறுக்க அனுமதிக்கிறது, பணி உருவாக்கம் மற்றும் தரவு கடந்து செல்வதை எளிதாக்குகிறது.
override(task_id=...)	ஒற்றை செயல்பாட்டிலிருந்து பல பணிகளை நிறுவும் போது ஒரு பணியின் ஐடியை மாறும் வகையில் மாற்றியமைக்கப் பயன்படுகிறது.
extract_elements(dag_run=None)	பணி செயல்படுத்தலை மாறும் வகையில் கட்டமைக்க DAG_RUN.CONF அகராதியிலிருந்து மதிப்புகளை பிரித்தெடுக்கும் ஒரு செயல்பாடு.
schedule_interval=None	ஒரு நிலையான அட்டவணையில் இயங்குவதற்குப் பதிலாக, கைமுறையாக தூண்டப்படும்போது மட்டுமே DAG செயல்படுத்தப்படுவதை உறுதி செய்கிறது.
op_args=[element]	பைதனோபரேட்டர் பணிக்கு மாறும் வாதங்களை அனுப்புகிறது, இது ஒரு பணி நிகழ்வுக்கு வெவ்வேறு மரணதண்டனைகளை செயல்படுத்துகிறது.
catchup=False	இடைநிறுத்தத்திற்குப் பிறகு தொடங்கும்போது தவறவிட்ட அனைத்து DAG மரணதண்டனைகளையும் இயக்குவதைத் தடுக்கிறது, இது நிகழ்நேர உள்ளமைவுகளுக்கு பயனுள்ளதாக இருக்கும்.

காற்றோட்டத்தில் இயக்க நேர உள்ளமைவுடன் டைனமிக் DAG களை உருவாக்குதல்

அப்பாச்சி காற்றோட்டம் என்பது சிக்கலான பணிப்பாய்வுகளைத் திட்டமிடுவதற்கான ஒரு சக்திவாய்ந்த கருவியாகும், ஆனால் அதன் உண்மையான வலிமை அதன் நெகிழ்வுத்தன்மையில் உள்ளது. முன்னர் வழங்கப்பட்ட ஸ்கிரிப்ட்கள் எவ்வாறு உருவாக்குவது என்பதை நிரூபிக்கின்றன டைனமிக் டாக் பணி சார்புநிலைகள் இயக்க நேரத்தில் தீர்மானிக்கப்படுகின்றன Dag_run.conf. செயலாக்க உறுப்புகளின் பட்டியலை ஹார்ட்கோடி செய்வதற்கு பதிலாக, தூண்டப்படும்போது DAG அவற்றை மாறும் வகையில் மீட்டெடுக்கிறது, மேலும் தழுவிக்கொள்ளக்கூடிய பணிப்பாய்வுகளை அனுமதிக்கிறது. மாறி தரவுத்தொகுப்புகளை செயலாக்குவது அல்லது வெளிப்புற நிலைமைகளின் அடிப்படையில் குறிப்பிட்ட பணிகளை செயல்படுத்துவது போன்ற நிஜ உலக சூழ்நிலைகளில் இது மிகவும் பயனுள்ளதாக இருக்கும். தினசரி மாற்றும் கோப்புகள் தினசரி மாற்றுவதற்கான ஒரு ஈடிஎல் பைப்லைனை கற்பனை செய்து பாருங்கள் - இந்த அணுகுமுறை ஆட்டோமேஷனை மிகவும் எளிதாக்குகிறது. .

முதல் ஸ்கிரிப்ட் பயன்படுத்துகிறது பைதோனோபரேட்டர் பணிகளைச் செயல்படுத்தவும், சார்புகளை மாறும் வகையில் அமைக்கவும். இது உறுப்புகள் பட்டியலைப் பிரித்தெடுக்கிறது dag_run.conf, தேவைப்படும்போது மட்டுமே பணிகள் உருவாக்கப்படுவதை உறுதி செய்தல். பட்டியலில் உள்ள ஒவ்வொரு உறுப்புகளும் ஒரு தனித்துவமான பணியாக மாறும், மேலும் சார்புகள் தொடர்ச்சியாக அமைக்கப்படுகின்றன. இரண்டாவது அணுகுமுறை டாஸ்க்ஃப்ளோ ஏபிஐ, இது அலங்காரக்காரர்களுடன் DAG உருவாக்கத்தை எளிதாக்குகிறது Ugdag மற்றும் Ask பணி. இந்த முறை DAG ஐ மேலும் படிக்கக்கூடியதாக ஆக்குகிறது மற்றும் தூய்மையான மரணதண்டனை தர்க்கத்தை பராமரிக்கிறது. குறியீடு மாற்றங்கள் தேவையில்லாமல் பணிப்பாய்வு வெவ்வேறு உள்ளமைவுகளுக்கு ஏற்ப மாற்ற முடியும் என்பதை இந்த அணுகுமுறைகள் உறுதி செய்கின்றன.

எடுத்துக்காட்டாக, ஒரு ஈ-காமர்ஸ் நிறுவனம் தொகுதிகளில் ஆர்டர்களை செயலாக்கும் ஒரு காட்சியைக் கவனியுங்கள். சில நாட்களில் மற்றவர்களை விட அதிக அவசர ஆர்டர்கள் இருக்கலாம், வெவ்வேறு பணி காட்சிகள் தேவைப்படுகின்றன. ஒரு நிலையான DAG ஐப் பயன்படுத்துவது என்பது ஒவ்வொரு முறையும் முன்னுரிமைகள் மாறும்போது குறியீட்டை மாற்றியமைப்பதைக் குறிக்கும். எங்கள் டைனமிக் டிஏஜி அணுகுமுறையுடன், ஒரு வெளிப்புற அமைப்பு ஒரு குறிப்பிட்ட பணி வரிசையுடன் DAG ஐத் தூண்டலாம், இதனால் செயல்முறையை மிகவும் திறமையாக மாற்றும். மற்றொரு பயன்பாட்டு வழக்கு தரவு அறிவியலில் உள்ளது, அங்கு உள்வரும் தரவு விநியோகங்களின் அடிப்படையில் மாதிரிகள் மறுபயன்பாடு தேவைப்படலாம். தேவையான மாதிரி உள்ளமைவுகளை மாறும் வகையில் அனுப்புவதன் மூலம், தேவையான கணக்கீடுகள் மட்டுமே செயல்படுத்தப்படுகின்றன, நேரத்தையும் வளங்களையும் மிச்சப்படுத்துகின்றன. .

சுருக்கமாக, இந்த ஸ்கிரிப்ட்கள் இயக்க நேர உள்ளீடுகளின் அடிப்படையில் DAG களை மாறும் வகையில் உருவாக்குவதற்கான ஒரு அடித்தளத்தை வழங்குகின்றன. அந்நியப்படுத்துவதன் மூலம் காற்றோட்டத்தின் பணிப்பாய்வு API அல்லது பாரம்பரிய பைதோனோபரேட்டர் அணுகுமுறை, டெவலப்பர்கள் நெகிழ்வான, மட்டு மற்றும் திறமையான பணிப்பாய்வுகளை உருவாக்க முடியும். இது கையேடு தலையீட்டின் தேவையை நீக்குகிறது மற்றும் பிற ஆட்டோமேஷன் அமைப்புகளுடன் தடையற்ற ஒருங்கிணைப்பை அனுமதிக்கிறது. வாடிக்கையாளர் ஆர்டர்களை செயலாக்குவது, தரவுக் குழாய்களை நிர்வகித்தல் அல்லது கிளவுட் பணிப்பாய்வுகளைத் தொடங்குவது, டைனமிக் DAG கள் குறிப்பிட்ட வணிகத் தேவைகளுக்கு ஏற்ப சிறந்த ஆட்டோமேஷனை இயக்குகின்றன.

இயக்க நேர உள்ளமைவுடன் காற்றோட்டத்தில் டைனமிக் டாஸ்க் வரிசைமுறையை செயல்படுத்துகிறது

அப்பாச்சி காற்றோட்டத்தைப் பயன்படுத்தி பைதான் அடிப்படையிலான பின்தளத்தில் ஆட்டோமேஷன்

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

மாற்று அணுகுமுறை: சிறந்த வாசிப்புக்கு டாஸ்க்ஃப்ளோ ஏபிஐ பயன்படுத்துதல்

ஏர்ஃப்ளோவின் பணிப்பாய்வு API ஐப் பயன்படுத்தி நவீன பைதான் அணுகுமுறை

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, இது முன் வரையறுக்கப்பட்ட தர்க்கத்தின் அடிப்படையில் செயல்படுத்த அடுத்த பணியை தீர்மானிக்கிறது. கோப்புகளை செயலாக்கும் ஒரு டைனமிக் டாக் எங்களிடம் உள்ளது என்று வைத்துக்கொள்வோம், ஆனால் ஒரு குறிப்பிட்ட அளவிற்கு மேலே உள்ள கோப்புகளுக்கு மட்டுமே சரிபார்ப்பு தேவைப்படுகிறது. எல்லா பணிகளையும் தொடர்ச்சியாக செயல்படுத்துவதற்கு பதிலாக, எந்த பணிகளை இயக்க வேண்டும், மரணதண்டனை நேரத்தை மேம்படுத்துதல் மற்றும் வள பயன்பாட்டைக் குறைத்தல் ஆகியவற்றை நாம் மாறும் என்பதை நாம் மாறும் தீர்மானிக்க முடியும். இந்த அணுகுமுறை தொடர்புடைய பணிப்பாய்வுகள் மட்டுமே தூண்டப்படுவதை உறுதி செய்கிறது, இதனால் தரவுக் குழாய்கள் மிகவும் திறமையாகின்றன. .

டைனமிக் DAG களை மேம்படுத்துவதற்கான மற்றொரு வழி இணைப்பதன் மூலம் XComs (குறுக்கு-தொடர்பு செய்திகள்). எக்ஸ்காம்கள் தரவைப் பரிமாறிக் கொள்ள பணிகளை அனுமதிக்கின்றன, அதாவது மாறும் வகையில் உருவாக்கப்பட்ட பணி வரிசை படிகளுக்கு இடையில் தகவல்களை அனுப்ப முடியும். எடுத்துக்காட்டாக, ஒரு ETL குழாய்த்திட்டத்தில், ஒரு முன் செயலாக்கப் பணி தேவையான மாற்றங்களைத் தீர்மானிக்கக்கூடும், மேலும் அந்த விவரங்களை அடுத்தடுத்த பணிகளுக்கு அனுப்பலாம். இந்த முறை உண்மையிலேயே தரவு சார்ந்த உந்துதல் பணிப்பாய்வுகளை செயல்படுத்துகிறது, அங்கு செயல்பாட்டு ஓட்டம் நிகழ்நேர உள்ளீடுகளின் அடிப்படையில் மாற்றியமைக்கிறது, ஆட்டோமேஷன் திறன்களை கணிசமாக அதிகரிக்கிறது.

காற்றோட்டத்தில் டைனமிக் பணி வரிசைமுறை பற்றிய பொதுவான கேள்விகள்

என்ன dag_run.conf பயன்படுத்தப்படுகிறது?
இது ஒரு DAG ஐத் தூண்டும் போது இயக்க நேரத்தில் உள்ளமைவு அளவுருக்களைக் கடந்து செல்ல அனுமதிக்கிறது, இது பணிப்பாய்வுகளை மிகவும் நெகிழ்வானதாக ஆக்குகிறது.
காற்றோட்டத்தில் பணிகளை நான் எவ்வாறு மாறும் வகையில் உருவாக்க முடியும்?
ஒரு பல நிகழ்வுகளை நிறுவ நீங்கள் ஒரு வளையத்தைப் பயன்படுத்தலாம் PythonOperator அல்லது பயன்படுத்தவும் @task பணிப்பாய்வு API இல் அலங்காரக்காரர்.
பயன்படுத்துவதன் நன்மை என்ன BranchPythonOperator?
இது நிபந்தனை செயல்படுத்தலை செயல்படுத்துகிறது, இது முன் வரையறுக்கப்பட்ட தர்க்கத்தின் அடிப்படையில் வெவ்வேறு பாதைகளைப் பின்பற்ற DAG களை அனுமதிக்கிறது, செயல்திறனை மேம்படுத்துகிறது.
எப்படி XComs டைனமிக் DAG களை மேம்படுத்தவா?
எக்ஸ்.காம்கள் தரவைப் பகிர பணிகளை அனுமதிக்கின்றன, அடுத்தடுத்த பணிகள் முந்தைய படிகளிலிருந்து பொருத்தமான தகவல்களைப் பெறுவதை உறுதிசெய்கின்றன.
நான் சார்புகளை மாறும் வகையில் அமைக்க முடியுமா?
ஆம், நீங்கள் பயன்படுத்தலாம் set_upstream() மற்றும் set_downstream() சார்புகளை ஒரு DAG க்குள் மாறும் வகையில் வரையறுக்கும் முறைகள்.

இயக்க நேர உள்ளமைவுகளுடன் டைனமிக் பணிப்பாய்வுகளை மேம்படுத்துதல்

செயல்படுத்துகிறது டைனமிக் பணி வரிசைமுறை காற்றோட்டத்தில் பணிப்பாய்வு ஆட்டோமேஷனை கணிசமாக மேம்படுத்துகிறது, இது மாறிவரும் தேவைகளுக்கு ஏற்றது. இயக்க நேர உள்ளமைவுகளை மேம்படுத்துவதன் மூலம், டெவலப்பர்கள் நிலையான DAG வரையறைகளைத் தவிர்க்கலாம், அதற்கு பதிலாக நெகிழ்வான, தரவு சார்ந்த குழாய்களை உருவாக்கலாம். நிதி அறிக்கை அல்லது இயந்திர கற்றல் மாதிரி பயிற்சி போன்ற நிகழ்நேர உள்ளீட்டின் அடிப்படையில் பணிகளை வரையறுக்க வேண்டிய சூழல்களில் இந்த அணுகுமுறை குறிப்பாக மதிப்புமிக்கது. .

ஒருங்கிணைப்பதன் மூலம் dag_run.conf, நிபந்தனை செயல்படுத்தல் மற்றும் சார்பு மேலாண்மை, குழுக்கள் அளவிடக்கூடிய மற்றும் திறமையான பணிப்பாய்வுகளை உருவாக்க முடியும். ஈ-காமர்ஸ் பரிவர்த்தனைகளை செயலாக்குவது, கிளவுட் அடிப்படையிலான தரவு மாற்றங்களை நிர்வகித்தல் அல்லது சிக்கலான தொகுதி வேலைகளைத் திட்டமிடுவது போன்றவை, காற்றோட்டத்தின் டைனமிக் டிஏஜி திறன்கள் உகந்த மற்றும் தானியங்கி தீர்வை வழங்குகின்றன. இந்த நுட்பங்களில் முதலீடு செய்வது கையேடு தலையீட்டைக் குறைக்கும் போது வணிகங்களை செயல்பாடுகளை நெறிப்படுத்த அனுமதிக்கிறது.

காற்றோட்டத்தில் டைனமிக் பணி வரிசைமுறைக்கான ஆதாரங்கள் மற்றும் குறிப்புகள்

அப்பாச்சி காற்றோட்டம் ஆவணங்கள் - DAG உள்ளமைவு மற்றும் இயக்க நேர அளவுருக்கள் பற்றிய விரிவான நுண்ணறிவு: அப்பாச்சி காற்றோட்டம் அதிகாரப்பூர்வ ஆவணங்கள்
டைனமிக் டாக் உருவாக்கம் குறித்த நடுத்தர கட்டுரை - பயன்படுத்துவதற்கான வழிகாட்டி Dag_run.conf டைனமிக் பணி வரிசைமுறைக்கு: நடுத்தர: காற்றோட்டத்தில் டைனமிக் டாக்ஸ்
ஸ்டேக் வழிதல் கலந்துரையாடல் - உள்ளீட்டு உள்ளமைவின் அடிப்படையில் மாறும் வகையில் DAG களை உருவாக்குவதற்கான சமூக தீர்வுகள்: வழிதல் நூலை அடுக்கி வைக்கவும்
தரவு பொறியியல் வலைப்பதிவு - அளவிடக்கூடிய காற்றோட்ட பணிப்பாய்வுகளை வடிவமைப்பதற்கான சிறந்த நடைமுறைகள்: தரவு பொறியியல் வலைப்பதிவு

DAG ரன் உள்ளமைவைப் பயன்படுத்தி காற்றோட்டத்தில் டைனமிக் பணி காட்சிகளை உருவாக்குதல்