排程每日工作
Data Engineering 入門
Vincent Vankrunkelsven
Data Engineer, DataCamp
到目前為止你已完成的事
- 使用
extract_course_data()與extract_rating_data()擷取 - 用
transform_fill_programming_language()清理 NA - 每門課的平均評分：
transform_avg_rating() - 取得可推薦的使用者與課程配對：
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 入門
