始める無料で始める

DAG を定義する

これまでの演習では、抽出、変換、ロードの各フェーズを個別に完成させてきました。今回はこれらすべてを、コンソールで確認できる1つの etl() 関数にまとめます。

etl() 関数は、関連するデータベースからコースと評価の生データを抽出し、破損したデータをクリーニングして欠損値を補完します。さらに、コースごとの平均評価を計算し、レコメンデーションを生成するための決定ルールに基づいておすすめコースを作成した上で、最終的にそのレコメンデーションをデータベースにロードします。

動画で見たように、etl() は db_engines という1つの引数を受け取ります。etl と db_engines はどちらもワークスペースで使用できるため、定義するタスクは一方をもう一方に渡して呼び出すだけで済みます。

この演習はコースの一部です

データエンジニアリング入門

コースを見る

演習の手順

  • DAG の定義を完成させ、毎日実行されるようにしましょう。その際、cron 記法を使用してください。
  • タスクを完成させ、データベースエンジンを渡して etl() 関数を呼び出すようにしましょう。

実践的なインタラクティブ演習

このサンプルコードを完成させて、この演習に挑戦してみましょう。

# Define the DAG so it runs on a daily basis
@dag(dag_id="recommendations",
     start_date=datetime(2024, 1, 1),
     schedule="____")
def recommendations():
    # Make sure the task calls etl() with the database engines
    @task(task_id="recommendations_task")
    def recommendations_task():
        ____(____)

    recommendations_task()

# Run the DAG
recommendations()
コードを編集して実行