Get startedGet started for free

Defining a DAG

In the previous exercises you applied the three steps in the ETL process:

  • Extract: Extract the film PostgreSQL table into pandas.
  • Transform: Split the rental_rate column of the film DataFrame.
  • Load: Load a the film DataFrame into a PostgreSQL data warehouse.

The functions extract_film_to_pandas(), transform_rental_rate() and load_dataframe_to_film() are defined in your workspace. In this exercise, you'll turn them into a scheduled Airflow pipeline: the etl task runs after a wait_for_table task signals the source table is ready.

This exercise is part of the course

Introduction to Data Engineering

View Course

Exercise instructions

  • Complete the etl() task by making use of the functions defined in the exercise description.
  • Set up the correct dependency. Note that the etl task should wait for wait_for_table to be finished.
  • The last line is a sample run: etl.function() calls the plain Python function behind the task, so the pipeline runs once here in the console.

Hands-on interactive exercise

Have a go at this exercise by completing this sample code.

# 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()
Edit and Run Code