Günlük işler zamanlama

Data Engineering'e Giriş

Vincent Vankrunkelsven

Data Engineer @ DataCamp

Şimdiye kadar yaptıklarınız

extract_course_data() ve extract_rating_data() ile çıkarın
transform_fill_programming_language() ile NA değerleri temizleyin
Ders başına ortalama puan: transform_avg_rating()
Uygun kullanıcı–ders id çiftleri: transform_courses_to_recommend()
Önerileri hesaplayın: transform_recommendations()

Postgres'e yükleme

Hesaplamaları veri ürünlerinde kullanın
Günlük güncelleyin
Örnek kullanım: öneri e-postaları göndermek

Yükleme aşaması

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

def etl(db_engines):
    # Veriyi çıkar
    courses = extract_course_data(db_engines)
    rating = extract_rating_data(db_engines)
    # Ders verisini temizle
    courses = transform_fill_programming_language(courses)

    # Ortalama ders puanlarını al
    avg_course_rating = transform_avg_rating(rating)

    # Uygun kullanıcı ve ders id çiftlerini al
    courses_to_recommend = transform_courses_to_recommend(
        rating,
        courses,
    )

    # Önerileri hesapla
    recommendations = transform_recommendations(
        avg_course_rating,
        courses_to_recommend,
    )

    # Önerileri veritabanına yükle
    load_to_dwh(recommendations, db_engine))

DAG oluşturma

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,
)

Hadi pratik yapalım!

Data Engineering'e Giriş