ડેગ રન ગોઠવણીનો ઉપયોગ કરીને એરફ્લોમાં ગતિશીલ કાર્ય સિક્વન્સ ઉત્પન્ન

એરફ્લોમાં ગતિશીલ કાર્ય અવલંબનની શક્તિને અનલ ocking ક કરવું

અપાચે એરફ્લો એ એક શક્તિશાળી વર્કફ્લો Auto ટોમેશન ટૂલ છે, પરંતુ ગતિશીલ અવલંબનને હેન્ડલ કરવું એ પઝલ હલ કરવા જેવું લાગે છે. નિર્દેશિત એસાયક્લિક ગ્રાફ (ડીએજી) ની રચના કરતી વખતે, હાર્ડકોડિંગ ટાસ્ક સિક્વન્સ સરળ ઉપયોગના કેસો માટે કાર્ય કરી શકે છે, પરંતુ જો રનટાઈમ પર માળખું નક્કી કરવાની જરૂર હોય તો શું? .

કલ્પના કરો કે તમે ડેટા પાઇપલાઇન પર કામ કરી રહ્યાં છો જ્યાં ચલાવવાનાં કાર્યો ઇનકમિંગ ડેટા પર આધારિત છે. ઉદાહરણ તરીકે, દૈનિક ગોઠવણીના આધારે ફાઇલોના વિવિધ સેટ્સ પર પ્રક્રિયા કરો અથવા વ્યવસાયના નિયમના આધારે ચલ પરિવર્તન ચલાવવું. આવા કિસ્સાઓમાં, સ્થિર ડીએજી તેને કાપશે નહીં - તમારે ગતિશીલતાપૂર્વક અવલંબનને વ્યાખ્યાયિત કરવાની રીતની જરૂર છે.

આ ચોક્કસપણે જ્યાં એરફ્લો છે dag_run.conf રમત-ચેન્જર હોઈ શકે છે. જ્યારે ડીએજીને ટ્રિગર કરતી વખતે રૂપરેખાંકન શબ્દકોશ પસાર કરીને, તમે ગતિશીલ રીતે કાર્ય સિક્વન્સ ઉત્પન્ન કરી શકો છો. જો કે, આને સ્ટ્રક્ચર્ડ રીતે અમલમાં મૂકવા માટે એરફ્લોના એક્ઝેક્યુશન મોડેલની deep ંડી સમજની જરૂર છે.

આ લેખમાં, અમે ગતિશીલ ડીએજી કેવી રીતે બનાવવી તે અન્વેષણ કરીશું જ્યાં કાર્ય અવલંબન રનટાઈમનો ઉપયોગ કરીને નક્કી કરવામાં આવે છે dag_run.conf. જો તમે આ પ્રાપ્ત કરવા માટે સંઘર્ષ કરી રહ્યાં છો અને સ્પષ્ટ સમાધાન મળ્યું નથી, તો ચિંતા કરશો નહીં - તમે એકલા નથી! ચાલો તેને વ્યવહારિક ઉદાહરણો સાથે પગલું દ્વારા પગલું ભરીએ. .

એરફ્લોમાં રનટાઇમ ગોઠવણી સાથે ગતિશીલ ડગ્સ બનાવવી

જટિલ વર્કફ્લોઝને ઓર્કેસ્ટ્રેટ કરવા માટે અપાચે એરફ્લો એક શક્તિશાળી સાધન છે, પરંતુ તેની સાચી શક્તિ તેની સુગમતામાં રહેલી છે. અગાઉ પ્રસ્તુત સ્ક્રિપ્ટો દર્શાવે છે કે કેવી રીતે બનાવવી ગતિશીલ ભીડ જ્યાં કાર્ય અવલંબનનો ઉપયોગ કરીને રનટાઈમ સમયે નક્કી કરવામાં આવે છે dag_run.conf. પ્રક્રિયા કરવા માટે તત્વોની સૂચિને હાર્ડકોડ કરવાને બદલે, જ્યારે વધુ સ્વીકાર્ય વર્કફ્લોની મંજૂરી આપે છે ત્યારે ડીએજી ગતિશીલ રીતે તેમને પ્રાપ્ત કરે છે. આ ખાસ કરીને વાસ્તવિક-વિશ્વના દૃશ્યોમાં ઉપયોગી છે, જેમ કે વેરીએબલ ડેટાસેટ્સ પ્રોસેસિંગ અથવા બાહ્ય પરિસ્થિતિઓના આધારે વિશિષ્ટ કાર્યો ચલાવવા. ઇટીએલ પાઇપલાઇનની કલ્પના કરો જ્યાં દરરોજ ફાઇલો પર ફેરફાર કરવાની પ્રક્રિયા કરવાની - આ અભિગમ ઓટોમેશનને વધુ સરળ બનાવે છે. .

પ્રથમ સ્ક્રિપ્ટનો ઉપયોગ કરે છે પાયગણો કરનાર કાર્યો ચલાવવા અને ગતિશીલતાપૂર્વક અવલંબન સેટ કરવા માટે. તે તત્વોની સૂચિ કા racts ે છે dag_run.conf, જ્યારે જરૂરી હોય ત્યારે કાર્યો બનાવવામાં આવે છે તેની ખાતરી કરવી. સૂચિમાં દરેક તત્વ એક અનન્ય કાર્ય બની જાય છે, અને અવલંબન ક્રમિક રીતે સેટ કરવામાં આવે છે. બીજો અભિગમ લાભ આપે છે ટાસ્કફ્લો API, જે સજાવટ સાથે ડેગ બનાવટને સરળ બનાવે છે @ડ ag ગ અને @ટાસ્ક. આ પદ્ધતિ ડીએજીને વધુ વાંચવા યોગ્ય બનાવે છે અને ક્લીનર એક્ઝેક્યુશન તર્ક જાળવી રાખે છે. આ અભિગમો સુનિશ્ચિત કરે છે કે વર્કફ્લો કોડ ફેરફારોની જરૂરિયાત વિના વિવિધ રૂપરેખાંકનોમાં અનુકૂળ થઈ શકે છે.

ઉદાહરણ તરીકે, એક દૃશ્યનો વિચાર કરો જ્યાં ઇ-ક ce મર્સ કંપની બ ches ચેસમાં ઓર્ડર પર પ્રક્રિયા કરે છે. કેટલાક દિવસોમાં અન્ય કરતા વધુ તાત્કાલિક ઓર્ડર હોઈ શકે છે, જેમાં વિવિધ કાર્ય સિક્વન્સની જરૂર હોય છે. સ્થિર ડીએજીનો ઉપયોગ કરીને દરેક વખતે પ્રાથમિકતાઓમાં ફેરફાર કરવા માટે કોડમાં ફેરફાર કરવો. અમારા ગતિશીલ ડીએજી અભિગમ સાથે, બાહ્ય સિસ્ટમ ચોક્કસ કાર્ય ક્રમ સાથે ડીએજીને ટ્રિગર કરી શકે છે, પ્રક્રિયાને વધુ કાર્યક્ષમ બનાવે છે. બીજો ઉપયોગ કેસ ડેટા સાયન્સમાં છે, જ્યાં મોડેલોને આવતા ડેટા વિતરણોના આધારે ફરીથી ગોઠવણીની જરૂર પડી શકે છે. ગતિશીલ રીતે જરૂરી મોડેલ રૂપરેખાંકનો પસાર કરીને, ફક્ત જરૂરી ગણતરીઓ ચલાવવામાં આવે છે, સમય અને સંસાધનોની બચત કરે છે. .

સારાંશમાં, આ સ્ક્રિપ્ટો રનટાઈમ ઇનપુટ્સના આધારે ગતિશીલ રીતે ડીએગ ઉત્પન્ન કરવા માટે પાયો પ્રદાન કરે છે. લાભ લઈને એરફ્લોનો ટાસ્કફ્લો API અથવા પરંપરાગત પાયથોનોપરેટર અભિગમ, વિકાસકર્તાઓ લવચીક, મોડ્યુલર અને કાર્યક્ષમ વર્કફ્લો બનાવી શકે છે. આ મેન્યુઅલ હસ્તક્ષેપની જરૂરિયાતને દૂર કરે છે અને અન્ય auto ટોમેશન સિસ્ટમ્સ સાથે સીમલેસ એકીકરણની મંજૂરી આપે છે. ગ્રાહકના ઓર્ડર પર પ્રક્રિયા કરે છે, ડેટા પાઇપલાઇન્સનું સંચાલન કરે છે, અથવા ક્લાઉડ વર્કફ્લોને c ર્કેસ્ટ્રેટ કરે છે, ગતિશીલ ડીએજીએસ વિશિષ્ટ વ્યવસાયની જરૂરિયાતોને અનુરૂપ સ્માર્ટ ઓટોમેશનને સક્ષમ કરે છે.

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, જે પૂર્વનિર્ધારિત તર્કના આધારે ચલાવવા માટે આગળનું કાર્ય નક્કી કરે છે. ધારો કે અમારી પાસે ગતિશીલ ડીએજી છે જે ફાઇલો પર પ્રક્રિયા કરે છે, પરંતુ ફક્ત ચોક્કસ કદથી ઉપરની ફાઇલોને માન્યતાની જરૂર હોય છે. ક્રમિક રીતે તમામ કાર્યોને અમલમાં મૂકવાને બદલે, અમે ગતિશીલ રીતે નક્કી કરી શકીએ કે કયા કાર્યો ચલાવવું, એક્ઝેક્યુશનનો સમય optim પ્ટિમાઇઝ કરવો અને સંસાધનનો ઉપયોગ ઘટાડવો. આ અભિગમ સુનિશ્ચિત કરે છે કે ફક્ત સંબંધિત વર્કફ્લો ટ્રિગર થાય છે, ડેટા પાઇપલાઇન્સને વધુ કાર્યક્ષમ બનાવે છે. .

ગતિશીલ ડગ્સને વધારવાની બીજી રીત એ શામેલ કરીને છે XComs (ક્રોસ-કમ્યુનિકેશન સંદેશા). એક્સકોમ કાર્યોને ડેટાની આપલે કરવાની મંજૂરી આપે છે, એટલે કે ગતિશીલ રીતે બનાવેલ કાર્ય ક્રમ પગલાં વચ્ચેની માહિતી પસાર કરી શકે છે. ઉદાહરણ તરીકે, ઇટીએલ પાઇપલાઇનમાં, પ્રીપ્રોસેસિંગ કાર્ય જરૂરી પરિવર્તન નક્કી કરી શકે છે અને તે વિગતોને અનુગામી કાર્યોમાં પસાર કરી શકે છે. આ પદ્ધતિ ખરેખર ડેટા-આધારિત વર્કફ્લોને સક્ષમ કરે છે, જ્યાં એક્ઝેક્યુશન ફ્લો રીઅલ-ટાઇમ ઇનપુટ્સના આધારે અનુકૂલન કરે છે, ઓટોમેશન ક્ષમતાઓમાં નોંધપાત્ર વધારો કરે છે.