Tạo chuỗi nhiệm vụ động trong luồng khí bằng cách sử dụng cấu hình DAG Run

Mở khóa sức mạnh của các phụ thuộc nhiệm vụ động trong luồng không khí

Apache Airflow là một công cụ tự động hóa dòng công việc mạnh mẽ, nhưng việc xử lý các phụ thuộc động đôi khi có thể cảm thấy như giải quyết một câu đố. Khi thiết kế một biểu đồ acyclic được định hướng (DAG), các chuỗi nhiệm vụ mã hóa cứng có thể hoạt động cho các trường hợp sử dụng đơn giản, nhưng nếu cấu trúc cần được xác định trong thời gian chạy thì sao? 🤔

Hãy tưởng tượng bạn làm việc trên một đường ống dữ liệu nơi các tác vụ được thực thi phụ thuộc vào dữ liệu đến. Ví dụ: xử lý các bộ tệp khác nhau dựa trên cấu hình hàng ngày hoặc thực hiện các biến đổi biến dựa trên quy tắc kinh doanh. Trong những trường hợp như vậy, một DAG tĩnh đã giành được nó đã cắt nó, bạn cần một cách để xác định các phụ thuộc một cách linh hoạt.

Đây chính xác là nơi luồng không khí dag_run.conf Có thể là một người thay đổi trò chơi. Bằng cách truyền từ điển cấu hình khi kích hoạt DAG, bạn có thể tạo tự động các chuỗi nhiệm vụ. Tuy nhiên, việc thực hiện điều này theo cách có cấu trúc đòi hỏi một sự hiểu biết sâu sắc về mô hình thực thi Airflow.

Trong bài viết này, chúng tôi sẽ khám phá cách xây dựng một DAG động trong đó các phụ thuộc nhiệm vụ được xác định trong thời gian chạy bằng cách sử dụng dag_run.conf. Nếu bạn đã phải vật lộn để đạt được điều này và không tìm thấy một giải pháp rõ ràng, thì đừng lo lắng, bạn không đơn độc! Hãy để chia nhỏ nó từng bước với các ví dụ thực tế. 🚀

Xây dựng DAG động với cấu hình thời gian chạy trong luồng khí

Apache Airflow là một công cụ mạnh mẽ để phối hợp các quy trình công việc phức tạp, nhưng sức mạnh thực sự của nó nằm ở sự linh hoạt của nó. Các tập lệnh được trình bày trước đó trình bày cách tạo Danh động DAG Trường hợp phụ thuộc nhiệm vụ được xác định trong thời gian chạy bằng cách sử dụng dag_run.conf. Thay vì mã hóa cứng danh sách các yếu tố cần xử lý, DAG lấy chúng một cách linh hoạt khi được kích hoạt, cho phép các quy trình công việc thích nghi hơn. Điều này đặc biệt hữu ích trong các kịch bản trong thế giới thực, chẳng hạn như xử lý bộ dữ liệu biến hoặc thực hiện các tác vụ cụ thể dựa trên các điều kiện bên ngoài. Hãy tưởng tượng một đường ống ETL nơi các tệp để xử lý thay đổi hàng ngày, cách tiếp cận này giúp tự động hóa dễ dàng hơn nhiều. 🚀

Tập lệnh đầu tiên sử dụng Pythonoperator Để thực hiện các tác vụ và đặt các phụ thuộc một cách linh hoạt. Nó trích xuất danh sách các yếu tố từ dag_run.conf, đảm bảo rằng các nhiệm vụ chỉ được tạo ra khi cần thiết. Mỗi yếu tố trong danh sách trở thành một nhiệm vụ duy nhất và các phụ thuộc được đặt tuần tự. Cách tiếp cận thứ hai tận dụng API Dòng Task, giúp đơn giản hóa sự sáng tạo của DAG với các nhà trang trí như @Dag Và @nhiệm vụ. Phương pháp này làm cho DAG dễ đọc hơn và duy trì logic thực hiện sạch hơn. Các phương pháp này đảm bảo rằng các quy trình công việc có thể thích ứng với các cấu hình khác nhau mà không cần thay đổi mã.

Ví dụ, hãy xem xét một kịch bản trong đó một công ty thương mại điện tử xử lý các đơn đặt hàng theo đợt. Một số ngày có thể có nhiều lệnh khẩn cấp hơn các đơn đặt hàng khác, yêu cầu các chuỗi nhiệm vụ khác nhau. Sử dụng DAG tĩnh có nghĩa là sửa đổi mã mỗi lần thay đổi. Với cách tiếp cận DAG động của chúng tôi, một hệ thống bên ngoài có thể kích hoạt DAG với một chuỗi nhiệm vụ cụ thể, làm cho quá trình này hiệu quả hơn. Một trường hợp sử dụng khác là trong khoa học dữ liệu, trong đó các mô hình có thể cần đào tạo lại dựa trên phân phối dữ liệu đến. Bằng cách chuyển các cấu hình mô hình cần thiết một cách linh hoạt, chỉ các tính toán cần thiết được thực thi, tiết kiệm thời gian và tài nguyên. 🎯

Tóm lại, các tập lệnh này cung cấp một nền tảng để tạo DAG động dựa trên các đầu vào thời gian chạy. Bằng cách tận dụng API dòng chảy của Airflow Hoặc cách tiếp cận Pythonoper truyền thống, các nhà phát triển có thể tạo ra các quy trình công việc linh hoạt, mô -đun và hiệu quả. Điều này giúp loại bỏ nhu cầu can thiệp thủ công và cho phép tích hợp liền mạch với các hệ thống tự động hóa khác. Cho dù xử lý các đơn đặt hàng của khách hàng, quản lý các đường ống dữ liệu hoặc phối hợp quy trình công việc đám mây, DAG động cho phép tự động hóa thông minh hơn phù hợp với các nhu cầu kinh doanh cụ thể.

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

Tăng cường trình tự nhiệm vụ động với thực thi có điều kiện trong luồng không khí

Một tính năng mạnh mẽ nhưng thường bị bỏ qua trong Apache Airflow là thực thi có điều kiện, có thể cải thiện hơn nữa tính linh hoạt của trình tự nhiệm vụ động. Trong khi lấy lại các phụ thuộc nhiệm vụ từ dag_run.conf là hữu ích, các kịch bản trong thế giới thực thường yêu cầu thực hiện chỉ các nhiệm vụ nhất định dựa trên các điều kiện cụ thể. Chẳng hạn, một số bộ dữ liệu có thể yêu cầu tiền xử lý trước khi phân tích, trong khi những bộ khác có thể được xử lý trực tiếp.

Thực thi có điều kiện trong luồng khí có thể được thực hiện bằng cách sử dụng BranchPythonOperator, trong đó xác định nhiệm vụ tiếp theo để thực thi dựa trên logic được xác định trước. Giả sử chúng ta có một DAG động xử lý các tệp, nhưng chỉ các tệp trên một kích thước nhất định yêu cầu xác thực. Thay vì thực hiện tất cả các tác vụ theo tuần tự, chúng ta có thể tự động quyết định các nhiệm vụ nào sẽ chạy, tối ưu hóa thời gian thực hiện và giảm sử dụng tài nguyên. Cách tiếp cận này đảm bảo rằng chỉ có các quy trình công việc có liên quan được kích hoạt, làm cho các đường ống dữ liệu hiệu quả hơn. 🚀

Một cách khác để tăng cường DAG động là kết hợp XComs (Thông điệp giao tiếp chéo). XCOM cho phép các tác vụ trao đổi dữ liệu, có nghĩa là một chuỗi nhiệm vụ được tạo động có thể truyền thông tin giữa các bước. Ví dụ, trong một đường ống ETL, một nhiệm vụ tiền xử lý có thể xác định các phép biến đổi cần thiết và chuyển các chi tiết đó cho các tác vụ tiếp theo. Phương pháp này cho phép các quy trình công việc thực sự dựa trên dữ liệu, trong đó luồng thực thi điều chỉnh dựa trên đầu vào thời gian thực, tăng khả năng tự động hóa đáng kể.