การกำหนดเวลารันงานประจำวัน

Data Engineering เบื้องต้น

Vincent Vankrunkelsven

Data Engineer, DataCamp

สิ่งที่ทำมาแล้ว

ดึงข้อมูลด้วย extract_course_data() และ extract_rating_data()
ล้างค่า NA ด้วย transform_fill_programming_language()
หาค่าเฉลี่ยคะแนนรีวิวต่อคอร์ส: transform_avg_rating()
หาคู่ user id และ course id ที่เข้าเงื่อนไข: transform_courses_to_recommend()
คำนวณคำแนะนำ: transform_recommendations()

การโหลดข้อมูลเข้าสู่ Postgres

นำผลการคำนวณไปใช้ในผลิตภัณฑ์ข้อมูล
อัปเดตทุกวัน
ตัวอย่างการใช้งาน: ส่งอีเมลพร้อมคำแนะนำ

ขั้นตอนการโหลดข้อมูล

recommendations.to_sql(
    "recommendations",
    db_engine,
    if_exists="append",
)

def etl(db_engines):
    # Extract the data
    courses = extract_course_data(db_engines)
    rating = extract_rating_data(db_engines)
    # Clean up courses data
    courses = transform_fill_programming_language(courses)

    # Get the average course ratings
    avg_course_rating = transform_avg_rating(rating)

    # Get eligible user and course id pairs
    courses_to_recommend = transform_courses_to_recommend(
        rating,
        courses,
    )

    # Calculate the recommendations
    recommendations = transform_recommendations(
        avg_course_rating,
        courses_to_recommend,
    )

    # Load the recommendations into the database
    load_to_dwh(recommendations, db_engines)

การสร้าง DAG

from airflow.sdk import dag, task

@dag(dag_id="recommendations",
     start_date=datetime(2024, 1, 1),
     schedule="0 0 * * *")
def recommendations():

    @task(task_id="recommendations_task")
    def recommendations_task():
        etl(db_engines)

    recommendations_task()

recommendations()

มาฝึกกันเถอะ!

Data Engineering เบื้องต้น