Putting it all together

1. Putting it all together

So we've now covered the full extent of an ETL pipeline. We've extracted data from databases, transformed the data to fit our needs, and loaded them back into a database, the data warehouse. This kind of batched ETL needs to run at a specific moment, and maybe after we completed some other tasks. It's time to put everything together.

2. The ETL function

First of all, it's nice to have your ETL behavior encapsulated into a clean `etl()` function. Let's say we have a `extract_table_to_df()` function, which extracts a PostgreSQL table into a pandas DataFrame. Then we could have one or many transformation functions that takes a pandas DataFrame and transform it by putting the data in a more suitable format for analysis. This function could be called `split_columns_transform()`, for example. Last but not least, a `load_df_into_dwh()` function loads the transformed data into a PostgreSQL database. We can define the resulting `etl()` function as follows. The result of `extract_table_to_df()` is used as an input for the transform function. We then use the output of the transform as input for `load_df_into_dwh`.

3. Airflow refresher

Now that we have a python function that describes the full ETL, we need to make sure that this function runs at a specific time. Before we go into the specifics, let's look at a small recap of Airflow. Apache Airflow is a workflow scheduler written in Python. You can represent directed acyclic graphs in Python objects. DAGs lend themselves perfectly to manage workflows, as there can be a dependency relation between tasks in the DAG. A task is a unit of work. Recall from chapter two that the quickest way to define one is to decorate a Python function with the task decorator. Airflow also ships a long list of ready-made operators for work that isn't a Python function, like running a bash script or waiting for a file to land, and you can write your own.

4. Scheduling with DAGs in Airflow

So the first thing we need to do is to create the DAG itself. In this code sample, we keep it simple and decorate a function called sample with the dag decorator. The argument that decides when the DAG runs is called `schedule`. There are multiple ways of defining it, but the most common one is using a cron expression. That is a string which represents a set of times. It's a string containing 5 characters, separated by a space. The leftmost character describes minutes, then hours, day of the month, month, and lastly, day of the week. Going into detail would drive us too far, but there are great resources online for learning cron expressions, and you'll find one linked on the slide. The DAG in the code sample runs every day at midnight.

5. The DAG definition file

Having created the DAG, it's time to set the ETL into motion. The etl() function we defined earlier is a plain Python function, so all it takes to turn it into a task is the task decorator. We give it a `task_id`, which is the name we'll see in the interface. Inside the DAG function, we call the task and connect it to whatever has to happen first: here an `EmptyOperator` standing in for a task that waits for the source table. The shift operator does the wiring: `etl` runs after `wait_for_table` completes. And notice we never had to tell the task which DAG it belongs to. Defining it inside the DAG function is enough.

6. The DAG definition file

Once you have this DAG definition and some tasks that relate to it, you can write it into a python file and place it in the DAG folder of Airflow. The service detects the DAG and shows it in the interface. One thing to watch for: that call to the DAG function at the bottom of the file is what registers it. Leave it out and Airflow will quietly show you nothing.

7. Airflow UI

The Airflow UI will look something like on the following screenshot. Note the name of the DAG and its schedule in the interface.

8. Let's practice!

That's all for now. Let's do some exercises.

Create Your Free Account

By continuing, you accept our Terms of Use, our Privacy Policy and that your data is stored in the USA.