일일 작업 스케줄링

데이터 엔지니어링 입문

Vincent Vankrunkelsven

Data Engineer @ DataCamp

지금까지 수행한 작업

extract_course_data()와 extract_rating_data()로 추출
transform_fill_programming_language()로 NA 정리
강좌별 평균 평점: transform_avg_rating()
추천 대상 사용자·강좌 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_engine))

DAG 생성

from airflow.models import DAG
from airflow.operators.python_operator import PythonOperator

dag = DAG(dag_id="recommendations",
          scheduled_interval="0 0 * * *")

task_recommendations = PythonOperator(
    task_id="recommendations_task",
    python_callable=etl,
)

연습해 봅시다!

데이터 엔지니어링 입문