Kích hoạt một Dag con
Bạn nhận thấy một số workflow đang dùng những thành phần tương tự và nhận ra bạn có thể tách các tác vụ dùng chung thành một Dag riêng. Cách này cho phép bạn chạy các thành phần đó khi cần mà không phải duy trì nhiều bản sao. Bạn quyết định thực thi một Dag con thông qua một task trong workflow hiện tại.
Các thành phần dag, task, và datetime đã được import sẵn cho bạn.
Bài tập này là một phần của khóa học
Giới thiệu về Apache Airflow bằng Python
Hướng dẫn bài tập
- Import operator cần thiết để khởi động một Dag từ trong workflow của bạn.
- Thiết lập operator để kích hoạt Dag có tên
child_pipeline. - Đảm bảo Dag cha chờ Dag đã kích hoạt hoàn tất trước khi tiếp tục.
- Thiết lập tần suất operator kiểm tra xem Dag con đã hoàn thành hay chưa.
Bài tập tương tác thực hành trực tiếp
Hãy thử làm bài tập này bằng cách hoàn thành đoạn mã mẫu này.
# Import TriggerDagRunOperator
from airflow.providers.standard.operators.trigger_dagrun import ____
@dag(start_date=datetime(2026, 1, 1))
def parent_orchestrator_dag():
# Trigger child_pipeline and wait for it to complete
trigger_child = TriggerDagRunOperator(
task_id="trigger_child_pipeline",
trigger_dag_id="____",
____=True,
____=30,
conf={"source": "s3://my-bucket/raw/"})
validate() >> trigger_child >> post_trigger_summary()
parent_orchestrator_dag()