DAG RUN yapılandırmasını kullanarak hava akışında dinamik görev dizileri üretme

Hava akışındaki dinamik görev bağımlılıklarının gücünün kilidini açmak

Apache Airflow güçlü bir iş akışı otomasyon aracıdır, ancak dinamik bağımlılıkların işlenmesi bazen bir bulmacayı çözmek gibi hissedebilir. Yönlendirilmiş bir asiklik grafik (DAG) tasarlarken, sabit kodlama görev dizileri basit kullanım durumları için işe yarayabilir, ancak yapının çalışma zamanında belirlenmesi gerekiyorsa? 🤔

Yürütülecek görevlerin gelen verilere bağlı olduğu bir veri boru hattı üzerinde çalıştığınızı düşünün. Örneğin, günlük yapılandırmaya dayalı farklı dosya kümelerinin işlenmesi veya bir iş kuralına dayalı değişken dönüşümleri yürütmek. Bu gibi durumlarda, statik bir DAG onu kesmez - bağımlılıkları dinamik olarak tanımlamanın bir yoluna ihtiyacınız vardır.

Bu tam olarak hava akışının Dag_run.conf bir oyun değiştirici olabilir. Bir DAG'yı tetiklerken bir yapılandırma sözlüğünü ileterek görev dizilerini dinamik olarak oluşturabilirsiniz. Bununla birlikte, bunu yapılandırılmış bir şekilde uygulamak, hava akışının yürütme modelini derinlemesine anlamayı gerektirir.

Bu makalede, görev bağımlılıklarının çalışma zamanında belirlendiği dinamik bir DAG'ın nasıl oluşturulacağını araştıracağız. Dag_run.conf. Bunu başarmak için mücadele ediyorsanız ve net bir çözüm bulamadıysanız, endişelenmeyin - yalnız değilsiniz! Pratik örneklerle adım adım yıkalım. 🚀

Airflow'da çalışma zamanı yapılandırması ile dinamik deg'ler oluşturma

Apache Airflow, karmaşık iş akışlarını düzenlemek için güçlü bir araçtır, ancak gerçek gücü esnekliğinde yatmaktadır. Daha önce sunulan senaryolar, dinamik dag Görev bağımlılıklarının çalışma zamanında kullanıldığı yerlerde Dag_run.conf. DAG, işlenecek öğelerin listesini sabitlemek yerine, tetiklendiğinde onları dinamik olarak alır ve daha uyarlanabilir iş akışlarına izin verir. Bu, özellikle değişken veri kümelerinin işlenmesi veya dış koşullara dayalı belirli görevlerin yürütülmesi gibi gerçek dünya senaryolarında kullanışlıdır. Dosyaların günlük olarak değiştirildiği bir ETL boru hattı hayal edin - bu yaklaşım otomasyonu daha kolay hale getirin. 🚀

İlk senaryo, Pitonoperatör Görevleri yürütmek ve bağımlılıkları dinamik olarak ayarlamak. Elementler listesini şuradan çıkarır Dag_run.conf, görevlerin yalnızca gerektiğinde oluşturulmasını sağlamak. Listedeki her öğe benzersiz bir görev haline gelir ve bağımlılıklar sırayla ayarlanır. İkinci yaklaşım, Taskflow API, DAG yaratmayı gibi dekoratörlerle basitleştiren @DAG Ve @görev. Bu yöntem DAG'ı daha okunabilir hale getirir ve daha temiz yürütme mantığını korur. Bu yaklaşımlar, iş akışlarının kod değişiklikleri gerektirmeden farklı yapılandırmalara uyum sağlayabilmesini sağlar.

Örneğin, bir e-ticaret şirketinin emirleri partiler halinde işlediği bir senaryoyu düşünün. Bazı günlerde farklı görev dizileri gerektiren diğerlerinden daha acil siparişlere sahip olabilir. Statik bir DAG kullanmak, öncelikler değiştiğinde kodun değiştirilmesi anlamına gelir. Dinamik DAG yaklaşımımızla, harici bir sistem DAG'ı belirli bir görev dizisiyle tetikleyebilir ve bu da süreci daha verimli hale getirebilir. Başka bir kullanım durumu, modellerin gelen veri dağıtımlarına dayanarak yeniden eğitilmesine ihtiyaç duyabileceği veri biliminde yer almaktadır. Gerekli model yapılandırmalarını dinamik olarak geçirerek, zaman ve kaynak tasarrufu sağlayarak yalnızca gerekli hesaplamalar yürütülür. 🎯

Özetle, bu komut dosyaları çalışma zamanı girişlerine dayanan dinamik olarak DAG'lar oluşturma temelleri sağlar. Kaldırarak Airflow'un Taskflow API'si Veya geleneksel pitonoperatör yaklaşımı, geliştiriciler esnek, modüler ve verimli iş akışları yaratabilir. Bu, manuel müdahale ihtiyacını ortadan kaldırır ve diğer otomasyon sistemleriyle sorunsuz entegrasyon sağlar. Müşteri siparişlerini işleme, veri boru hatlarını yönetme veya bulut iş akışlarının düzenlenmesi olsun, dinamik DAG'lar belirli iş ihtiyaçlarına göre uyarlanmış daha akıllı otomasyonu etkinleştirir.

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

Hava akışında koşullu yürütme ile dinamik görev sıralamasının geliştirilmesi

Güçlü ama sık sık gözden kaçan bir özellik Apache hava akışı Dinamik görev sıralamasının esnekliğini daha da artırabilen koşullu yürütülmedir. Görev bağımlılıklarını alırken Dag_run.conf yararlıdır, gerçek dünya senaryoları genellikle belirli koşullara göre yalnızca belirli görevlerin yürütülmesini gerektirir. Örneğin, bazı veri kümeleri analizden önce önceden işleme gerektirebilirken, diğerleri doğrudan işlenebilir.

Hava akışında koşullu yürütme kullanılarak uygulanabilir BranchPythonOperator, önceden tanımlanmış mantığa dayalı olarak yürütülecek bir sonraki görevi belirler. Dosyaları işleyen dinamik bir DAG'ımız olduğunu varsayalım, ancak yalnızca belirli bir boyutun üzerindeki dosyalar doğrulama gerektirir. Tüm görevleri sırayla yürütmek yerine, hangi görevlerin çalıştırılacağına, yürütme süresini optimize edeceğine ve kaynak kullanımını azaltan dinamik olarak karar verebiliriz. Bu yaklaşım, veri boru hatlarını daha verimli hale getirerek yalnızca ilgili iş akışlarının tetiklenmesini sağlar. 🚀

Dinamik DAG'ları geliştirmenin başka bir yolu, XComs (Çapraz iletişim mesajları). Xcoms, görevlerin veri alışverişine izin verir, yani dinamik olarak oluşturulan bir görev sırası adımlar arasında bilgi aktarabilir. Örneğin, bir ETL boru hattında, bir ön işleme görevi gerekli dönüşümleri belirleyebilir ve bu ayrıntıları sonraki görevlere aktarabilir. Bu yöntem, yürütme akışının gerçek zamanlı girişlere göre adapte olduğu ve otomasyon özelliklerini önemli ölçüde artırdığı gerçekten veri odaklı iş akışlarını mümkün kılar.