การสร้าง Task ด้วย Decorator

สมมติว่าต้องการสร้าง Airflow task ด้วย Python โดย task นี้จะดาวน์โหลดและบันทึกไฟล์ลงในระบบผ่าน Airflow โดยใช้ฟังก์ชัน pull_file ซึ่งกำหนดไว้ให้แล้ว

ไลบรารี requests และ Airflow dag object ถูก import ไว้ให้เรียบร้อยแล้ว

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

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

ดูคอร์ส

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

Import object ที่จำเป็นจากไลบรารี Airflow
กำหนดให้ฟังก์ชัน pull_file เป็น Airflow task
เรียกใช้งาน task object

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

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

# Import the task decorator
from ____ import ____

@dag(dag_id='file_update')
def file_update():
  # Decorate the function as a task
  ____
  def pull_file(URL, savepath):
    r = requests.get(URL)
    with open(savepath, 'wb') as f:
      f.write(r.content)   
    print(f"File pulled from {URL} and saved to {savepath}")
  # Call the task
  ____('http://dataserver/sales.json', 'latestsales.json')

file_update()

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