What is left of this stage is to build the main functions to populate the two schema layers for staging
and core.
This will bring all together the functions that we have defined for connections, inserts, updates,
deletions, and transformations.
To do this, we need to import the functions that we defined in the other scripts.
So let's first start by creating a new script.
Let's call it Datawarehouse.
And here we will do the imports.
We also need to import the logging and the airflow task decorator.
And same as we have done already, we start by setting up the logger and defining the global variable
for the table name.
And now we can proceed to the main function relating to the staging table.
So let's define a function called staging table.
That will create the actual staging table in our data warehouse.
And since we are working with airflow, we need to use the task decorator.
In the function, we define the schema variable and also initialize the connection and cursor.
So the schema will be staging and we define the quantum cursor as none.
Next we open a try and accept clause.
And in the try we start off by setting up the connection and cursor variables using the very first function
that we introduced in this section, which we got from the utils script.
We load the raw data, extract it from the YouTube API, and save this JSON file using the load data
function from the data loading script.
Here I'm realizing that the load data doesn't seem to have been imported correctly.
Now let's go into the loading, and we can see here that it actually was defined as load underscore
path.
So maybe let's change this to load data.
And now going back to the data warehouse script.
Exactly.
We should see that this will be showing as it has been properly imported.
We then create the schema and table for staging.
If they don't exist.
So create schema schema, create table with schema argument and also define a variable table IDs to
get all the video IDs for the staging table by using the get video IDs defined in utils.
Having these variables defined, we can now loop through each row and insert or update depending on
certain conditions.
So let's start the for loop for Follow in YouTube data.
If it's a first time insert, which means the table is empty.
We write if the length.
The table ids is equal to zero.
We insert data using the insert rows defined in data modification script.
Else if the table is already populated with data, we have one of two actions.
We either update the data of videos that are already present in the staging table, and if the video
ID in the loaded JSON is not in the staging table, that means we have a new video, so we need to insert
it into the staging table.
These two functions combined encapsulate an upsert logic.
So this covers the insertion and update part.
Now we just need to take care of possible removal of videos by the channel.
So we'll define a set variable called IDs in JSON, which will give all the video IDs that are present
in the JSON data.
We define a set by using the curly brackets.
Next we define an IDs to delete variable.
That will be the difference between the video IDs we have in the staging table and the above defined
IDs in JSON.
We can now convert the table IDs to a Python set, so we can use the minus operator to get the difference
between the two sets.
If there is any.
So this will look as follows.
A note on this is that we convert a table ids to a set, so we can use this inbuilt functionality of
Python sets to get the difference between two sets.
If I just delete is populated, then we use the delete rows function we defined in the data modification
script.
So if it is to delete exists.
The final step in the try clause is to log the table update so we can use logger.info.
If for some reason any part of the clause fails, we log this as an error and the except clause and
raise it.
As a final step, we use the finally block to ensure that the connection and cursor are closed so that
even if the try clause fails and we go in the accept, whatever happens, the code will still go inside
the finally clause and close both the connection and the character so we can write if they are not null
those anchors.
And that's it for the staging table function.
Now we have to do something quite similar for the core layer.
So let's get right into it.
Just as we did for staging, we need to define the core table function, which in this case will have
the name core table.
We will set the schema to core.
And we will also initialize the conn and cursor variable to null.
We can copy the first few lines of code that we have in the staging table function, also into the core
table function.
From here we need to initialize a variable, which we will call current video IDs that will be used
to collect all video IDs from the current execution.
This will be used for deletion logic and we can set this to an empty set.
Next, we execute a command to get all the data from the staging table and fetch all the rows.
As an important note here, as you can see, we are selecting all the data from the staging table.
It could very well be the case that you are dealing with tables that have hundreds of thousands of rows,
if not millions or even billions.
With these table sizes, this would not be a memory efficient approach.
So we would need to process the data in batches of a specific batch size, which you could set dependent
on your data size.
However, in our case, we are dealing with only a few hundreds of rows, so we will not work in batches
as it would be overengineering our problem.
From here it is again very similar to the staging table, as we define a for loop to go through each
row and transform or insert or update depending on certain conditions.
And as we loop through each row of data, we keep track of each video ID by appending it to the current
video IDs we defined earlier.
For the first insert in the core table, we check the table is empty and if yes, we transform the row
and insert it into the table.
As we saw in the staging main function, if the table is not empty, we populate the else clause as
follows.
We again use the transform data function to transform data that is already present on staging, and
then if the data is already present, we use the update function.
If not, then we have new data coming in, so we use the insert function.
Same as for the staging table.
We also take into account deletions.
So we define IDs to delete as the difference between what is already in the table, but not in the current
execution.
And finally, we delete the rows, if any are present using the delete rows function.
We then log the table update and define the accept clause for any errors we might have.
So let's copy and paste this from the staging, making sure we have the right tabs.
This has to be aligned with the try as a last step.
Same as we have done the staging function, we define the finally block so that regardless of what happens,
we close the connection and the cursor if they are not, none.
Having created these two main functions, what is left now is to simply define the airflow Dag related
to populating and updating data warehouse.
We will cover this in the next lecture.