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

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

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

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

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

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

காற்றோட்டத்தில் இயக்க நேர உள்ளமைவுடன் டைனமிக் 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])

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