การเพิ่ม Retry

คุณสังเกตเห็นว่า DAG หนึ่งเกิดข้อผิดพลาดบ่อยครั้งในงานที่ดึงข้อมูลจากแหล่งที่มาหนึ่ง และน่าหงุดหงิดที่การรันงานเดิมอีกครั้งหลังจากผ่านไปไม่กี่นาทีกลับแก้ปัญหาได้ หลังจากเรียนรู้เกี่ยวกับฟังก์ชัน Retry ใน Airflow DAG แล้ว จึงตัดสินใจนำ Retry มาใช้กับงานนี้เพื่อไม่ต้องรีสตาร์ทด้วยตนเองอีกต่อไป

dag, task, และ timedelta ถูก import มาให้แล้ว

แบบฝึกหัดนี้เป็นส่วนหนึ่งของหลักสูตร

Apache Airflow เบื้องต้นด้วย Python

ดูคอร์ส

คำแนะนำการฝึกหัด

ตั้งค่าให้งาน extract_data ลองใหม่ได้ 3 ครั้งก่อนที่จะล้มเหลว
เพิ่มระยะเวลารอ 10 นาทีระหว่างแต่ละครั้งที่ลองใหม่สำหรับงาน extract_data

แบบฝึกหัดเชิงโต้ตอบแบบลงมือทำ

ลองทำแบบฝึกหัดนี้โดยเติมโค้ดตัวอย่างนี้ให้สมบูรณ์

@dag(schedule="@daily", start_date=datetime(2026, 5, 1))
def etl_pipeline():

  # Set retries and retry delay on extract_data
  @task(____=3, ____=____(minutes=10))
  def extract_data():
    print("Extracting data from source...")
  
  @task()
  def process_source_data():
    print("Now processing data...")

  extract_data() >> process_source_data()
  
etl_pipeline()

แก้ไขและรันโค้ด