始める無料で始める

DAG の定義

これまでの演習では、抽出・変換・ロードの各フェーズを個別に行いました。今回は、それらをひとつにまとめた etl() 関数をコンソールで確認できるようになっています。

etl() 関数は、関連するデータベースからコースと評価の生データを抽出し、破損データのクレンジングや欠損値の補完を行い、コースごとの平均評価を計算し、レコメンデーションのための意思決定ルールに基づいて推奨を作成し、最後にそのレコメンデーションをデータベースにロードします。

動画で説明したとおり、etl() は 1 つの引数 db_engines を受け取ります。これは PythonOperator の op_kwargs に渡せます。呼び出し可能オブジェクトに対して、kwargs として展開される辞書を渡してください。

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

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

コースを見る

演習の手順

  • DAG の定義を完成させ、毎日実行されるようにしましょう。cron 記法を使ってください。
  • PythonOperator() に正しい引数を渡して完成させましょう。etl に加えて、db_engines もワークスペースで利用可能です。

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

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

# Define the DAG so it runs on a daily basis
dag = DAG(dag_id="recommendations",
          schedule_interval="____")

# Make sure `etl()` is called in the operator. Pass the correct kwargs.
task_recommendations = PythonOperator(
    task_id="recommendations_task",
    python_callable=____,
    op_kwargs={"____": ____},
)
コードを編集して実行