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() はワークスペースにすでに定義されています。この演習では、これらを使ってスケジュール実行される Airflow のパイプラインを構築します。etl タスクは、wait_for_table タスクが元テーブルの準備完了を知らせた後に実行されます。
この演習はコースの一部です
データエンジニアリング入門
演習の手順
- 演習の説明で定義されている関数を使って、
etl()タスクを完成させてください。 - 正しい依存関係を設定してください。
etlタスクはwait_for_tableの完了を待つ必要がある点に注意してください。 - 最後の行はサンプル実行です。
etl.function()はタスクの背後にあるプレーンな Python 関数を呼び出すため、ここではコンソール上で一度だけパイプラインが実行されます。
実践的なインタラクティブ演習
このサンプルコードを完成させて、この演習に挑戦してみましょう。
# Define the ETL task
@task(task_id="etl_film")
def etl():
film_df = ____()
film_df = ____(____)
____(____)
# Add the task to the DAG
@dag(dag_id="etl",
start_date=datetime(2024, 1, 1),
schedule="0 0 * * *")
def etl_dag():
wait_for_table = EmptyOperator(task_id="wait_for_table")
wait_for_table >> ____
etl_dag()
# Sample run
etl.function()