DAG を定義する
前の演習では、ETL プロセスの3つのステップを実行しました。
- Extract: PostgreSQL の
filmテーブルをpandasに取り出しました。 - Transform:
filmDataFrame のrental_rate列を分割しました。 - Load:
filmDataFrame を PostgreSQL のデータウェアハウスにロードしました。
extract_film_to_pandas()、transform_rental_rate()、load_dataframe_to_film() 関数はワークスペース内で定義済みです。この演習では、既存の DAG に ETL タスクを追加します。拡張する DAG と、待機すべきタスクはそれぞれ dag と wait_for_table としてワークスペース内に定義されています。
この演習はコースの一部です
データエンジニアリング入門
演習の手順
- 演習の説明で定義されている関数を使って、
etl()関数を完成させてください。 etl_taskがetlの callable を使用するようにしてください。- 正しい上流の依存関係を設定します。
etl_taskはwait_for_tableの完了を待つ必要があります。 - サンプルコードにはサンプル実行が含まれています。つまり、コードを実行すると ETL パイプラインが実行されます。
実践的なインタラクティブ演習
このサンプルコードを完成させて、この演習に挑戦してみましょう。
# Define the ETL function
def etl():
film_df = ____()
film_df = ____(____)
____(____)
# Define the ETL task using PythonOperator
etl_task = PythonOperator(task_id='etl_film',
python_callable=____,
dag=dag)
# Set the upstream to wait_for_table and sample run etl()
etl_task.____(wait_for_table)
etl()