การส่งข้อมูลด้วย XCom

คุณกำลังอัปเดต ETL workflow เพื่อใช้ XCom ในการติดตามข้อมูลได้สะดวกยิ่งขึ้น โดยจำเป็นต้องอัปเดต dependency ระหว่าง task ใน Dag ของคุณ

มี task ที่กำหนดไว้แล้ว 4 รายการ ดังนี้:

extract() - คืนค่า order_data
transform(order_data) - คืนค่าข้อมูลที่สรุปแล้ว
load(summary_data) - ไม่คืนค่าข้อมูล XCom
send_report() - task ที่ไม่ใช้ XCom สำหรับจัดการการแจ้งเตือน

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

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

ดูคอร์ส

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

แทรก task transform ระหว่าง extract() และ load() เพื่อให้ chain สมบูรณ์
กำหนดผลลัพธ์ที่ได้ให้กับตัวแปรชื่อ etl_result
ตั้งค่าให้ send_report รันหลังจากตัวแปร etl_result

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

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

@dag(start_date=datetime(2026,4,1))
def etl_example():
    # Chain extract, transform, and load, assigning the result
    ____ = load(____(extract()))
    # Run send_report after the ETL tasks
    etl_result ____ ____()
    
etl_example()

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