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

การกำหนด DAG

ในแบบฝึกหัดก่อนหน้า คุณได้ดำเนินการขั้นตอน extract, transform และ load แยกกันไปแล้ว ตอนนี้ทั้งหมดถูกรวมไว้ในฟังก์ชัน etl() เดียว ซึ่งสามารถดูได้ในคอนโซล

ฟังก์ชัน etl() จะดึงข้อมูลคอร์สและคะแนนดิบจากฐานข้อมูลที่เกี่ยวข้อง ทำความสะอาดข้อมูลที่เสียหายและเติมค่าที่หายไป คำนวณคะแนนเฉลี่ยของแต่ละคอร์ส และสร้างคำแนะนำตามกฎการตัดสินใจ จากนั้นโหลดคำแนะนำเหล่านั้นเข้าสู่ฐานข้อมูล

ตามที่ได้เรียนในวิดีโอ ฟังก์ชัน etl() รับอาร์กิวเมนต์เดียวคือ db_engines ซึ่งสามารถส่งค่านี้ไปยัง task ได้ผ่าน op_kwargs ใน PythonOperator โดยส่งเป็น dictionary ที่จะถูกนำไปใช้เป็น kwargs ใน callable

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

Data Engineering เบื้องต้น

ดูคอร์ส

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

  • กำหนด DAG ให้ครบถ้วน โดยตั้งให้รันทุกวัน อย่าลืมใช้รูปแบบ cron notation
  • กำหนดค่าใน PythonOperator() ให้ครบถ้วนโดยส่งอาร์กิวเมนต์ที่ถูกต้อง นอกจาก etl แล้ว ยังมี db_engines อยู่ใน workspace ของคุณด้วย

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

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

# Define the DAG so it runs on a daily basis
dag = DAG(dag_id="recommendations",
          schedule_interval="____")

# Make sure `etl()` is called in the operator. Pass the correct kwargs.
task_recommendations = PythonOperator(
    task_id="recommendations_task",
    python_callable=____,
    op_kwargs={"____": ____},
)
แก้ไขและรันโค้ด