เริ่มต้นใช้งานเริ่มต้นใช้งานได้ฟรี

กำหนด @task.branch

หลังจากเรียนรู้เรื่อง conditional logic ใน Airflow แล้ว คุณอยากทดลองใช้ decorator @task.branch โดยต้องการให้รัน code path ที่ต่างกัน หากวันที่ execution ปัจจุบันเป็นปีใหม่ (เช่น 2026 เทียบกับ 2025)

มีการกำหนด Dag พร้อม task ที่เกี่ยวข้องให้แล้ว หน้าที่ของคุณตอนนี้คือ implement @task.branch

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

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

ดูคอร์ส

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

  • เพิ่ม template variable วันที่ของ Airflow ที่เหมาะสมเป็น parameter ของ year_check เพื่อเปรียบเทียบปีของ execution ปัจจุบันกับปีก่อนหน้า
  • กำหนด current_year และ previous_year โดย slice 4 ตัวอักษรแรกจาก parameter วันที่แต่ละตัว
  • กำหนด dependency ให้กับ current_year_task และ new_year_task

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

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

@dag(start_date=datetime(2026,5,1), schedule='@monthly')
def process_yearly_expenses():
  # Create a function to determine if years are different
  @task.branch
  def year_check(____, ____):
      current_year = int(____[0:4])
      previous_year = int(____[0:4])
      if current_year == previous_year:
          return 'current_year_task'
      else:
          return 'new_year_task'

  # Define the dependencies
  branch_task __ current_year_task
  ____ >> new_year_task
แก้ไขและรันโค้ด