安排每日作业
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 入门
