Going back to the videos that script has created in the data extraction section, we can modify this
code as you see here, to create an airflow Dag.
What we first need to do is to put this script under the DAGs folder, so that it will be mounted to
the Docker containers where airflow is running.
Before doing so, however, create another subfolder and call it API.
Once you do that, you can move the videos that script under the DAGs folder.
Just note I want to make is that this DAGs folder together with the config include logs, plugins and
tests.
They were created in the previous lecture when we had run the docker compose up command.
Now what we need is a way to turn our Python code into airflow tasks.
This is done by using Python decorators.
We will not get too much in detail about Python decorators, but just know that the creators are a way
to wrap a function within another function.
And this wrapping effectively extends the functionality of the base function in order to define an airflow
task decorator.
You would use the at symbol so it would look something like this, followed by the task decorator.
Also, to use this task decorator we need to import it from airflow.
So let's do it right now from airflow dot decorators import task.
Now this decorator creation we need to do it for each of the functions that we have in the script.
Another thing is we need to change in the code is how we are handling the sensitive variables like the
API key and nonsensitive variables like the channel handle.
As we all know these variables, we define them already in the EMV and by referencing them in the docker
compose using the airflow via underscore variable name notation.
This can be seen in the Managing Variables document in the airflow docs, and we can take the example
under this section.
As you can see, it is using the airflow underscore underscore variable name notation.
So coming back to VSCode to ensure that these airflow variables are being defined properly you can use
the Docker exec command.
So let's do that right now.
And let's say that we will use the airflow web server as an example.
The first thing that we need to do is to check if the Docker containers are running.
Remember if Docker PS gives nothing back, it means you stopped your Docker services.
So you would need to run the docker compose up minus d command.
I will not do that.
Instead I will write the docker exec minus it.
We get the airflow web server service name and then bash.
From here we can type the grep command.
To check if we have variables starting with airflow var at the beginning.
As you can see, the two variables for the API key and channel have been returned.
Now that we confirmed the environment variables are present, we can simply call them using the variable
dot get syntax.
So these two commands over here will need to replace with the variable syntax.
And they can be rewritten as variable dot get and then the API key.
Same thing for the channel handle.
We can also filter out and the view and also remove these three lines of code.
And also very important we need to import the variable class.
So let's do that right now from airflow dot believe it's under models import variable.
One thing to note is that the variables won't show up using the airflow variables list command.
Let's showcase that right now.
Variables list.
As you can see, it returned a no data found.
This is because the variables are defined as environment variables and the same would apply for the
UI.
So we had to go on the airflow UI under the variable section.
We will find these variables.
And perhaps you can demonstrate this right now.
So the airflow web server is up and running.
Let's go here.
So this is the airflow UI.
If you recall we had defined the username and password which is airflow.
And same thing 1234.
So let's check the variable section under admin.
And as you can see there are no variables listed here.
Let's go back to VS code.
Now in terms of changes to the functions and inputs in the script, we are done.
However, we still need to create the actual Dag.
Apart from that, we also need to define the tasks and set dependencies.
For this, it is a general best practice that you create a Main.py script in the DAGs folder that will
contain our Dag code.
So let me do that right now.
Define main dot pi.
Okay.
It's important that this main square y is not inside the API subfolder, but it is under the general
dex folder.
We will create other subfolders as we go along in this course.
This may not be why we can populate by first doing the needed inputs.
So let's start from airflow import Dag.
This imports the main Dag class.
We also import pendulum.
The pendulum module is for dealing with time zones.
And since we also need to deal with DateTime and Timedelta classes, we need to import the date, time
and time delta modules.
And also need to import the functions in the videos that Pi script.
For now we will only have one leg, but as we progress we will create more.
These legs will have some common arguments.
So in order to factor this in, airflow has a default args argument that we can specify in the Dag build
and default args.
I already have one that we can use, which is this one over here.
Going over it briefly here we have an example of default arguments where we can specify things like
the Dag owner, what to do if emails are sent for failures, how many retries of the deck will occur
if it fails, and more.
In the start and end date arguments, which is the period defined for when the Dag will run, you see
this Local underscore time zone variable where you can specify your local time zone, which for my home
country is Malta.
This you can see it here.
You can adjust this depending where you are watching this course.
You can just search for time zone database and you will find many websites that have this list.
The page you see here is one of them and I will include it in the appendix of the section.
Simply use the value under the Time Zone column.
For Malta it is Europe Malta as you see it here.
Going back to the script, something which I would like to point out, which confused me a bit when
I first started using airflow is the start time.
The starting airflow is the time at which airflow will begin running the Dag, but the first run will
be scheduled at the end of the interval following the start date.
So, for example, we have a deck that runs daily, and we set the start date to the 1st of January
of 2025 at midnight.
the next day.
Grande will be on the 2nd of January at midnight, which is the end of the first interval following
the start date.
This is one of those things which might be counterintuitive at first, and I just wanted to mention
it, as it is good to know when defining the start dates and intervals.
Now that we have these details out of the way, we can focus on the actual creation.
Let's go to the DAGs airflow docs, and here you will see that there are three ways to actually declare
a Dag.
The first one is using a with statement.
It's the one you see here.
The next one is the standard constructor, which is this one here.
And the last one is using the Dag decorator, which is this one here.
As a matter of personal preference, I prefer using the with context manager.
Going back to VS code, we can start writing up our Dag.
The first important argument that we need to define is the dag id.
I will name it as produce JSON.
And why I chose this name is simple.
Ultimately, this Dag needs to output a JSON file under the data directory for default arts.
We can use the default args we have here.
We can also include a description which I recommend that you do, especially as the number of DAGs will
continue to grow.
I will say this Dag to produce JSON file with the raw data.
We also specify the schedule.
Let's say in this case that it will run every day at 2 p.m..
So in cron format this would look like zero 14.
Star star and star.
If you want to confirm this, you can go on Crontab Guru and replace with the values that we just used.
As you can see it is a daily run at 2:00 pm.
Finally, we can specify that catch up equals false.
What this does is it tells airflow not to catch up on missed diagrams from the past.
So let's conclude this with as Dag.
And here we can define the tasks by calling the functions in the videos task script.
So let's define tasks.
We start off with the playlist id.
Playlist id equals get playlist id.
The actual function video IDs we get from the video id function referencing the playlist ID.
Extracted data we get from the extracted video data function with the video IDs argument.
And finally, we save the JSON using the safety JSON function with the extracted data argument.
The last thing that we can do at this stage is to define dependencies.
When we talk about dependencies, this refers to, in what order will the tasks run from left to right?
This is important since the video IDs, for example, needs the output from the channel params to run.
Before that, let me just simply change this to underscore task so it doesn't get confused in terms
of conflict with the actual function name.
And we can write the task dependencies.
So playlist id video IDs.
Extract the data and save JSON first.
So here we are saying that this data JSON task depends on extracted data.
This depends on video IDs.
And the first thing it's going to run is the playlist ID function.
And that's it.
Congratulations you have built your first airflow Dag.
Now if the local time was 2 p.m..
This deck would run automatically if it is active.
However, we can also manually trigger the deck, and an easy way to do this is to go on the airflow
UI.
So let's go back to the airflow UI.
We run a refresh.
So it seems that we have an error with our deck.
And I believe I know why.
So this should actually be the environment variable name and not the value.
So let's go back to VS code and change this to channelhandler.
Let's refresh again.
Now as you can see the refresh of dishes we had and we have the produce underscore JSON deck we just
built.
If we are to actually activate this deck, we would simply just need to run the trigger deck button
here.
From here, let's go inside the Dag itself and see what functionalities airflow provides.
So now that we have clicked on the Dag itself, we will go over the main Sections that I would say irrelevant.
Here you have the detail section where you have a brief overview of the parameters of your DAC.
Here you get to see the DAC lineage, and here you get to see also the actual code that makes up your
DAC.
Going back to the actual DAGs, you will see that currently the Dag itself is paused.
This is the intended behavior as that is how we defined it in the docker compose.
So this is something that is by default, but you can change it inside the docker compose.
So what we can do right now to actually test out our Dag is to compose our Dag going inside.
And you will start seeing the execution of the Dag itself.
This dark green means that the actual tasks complete successfully, while the light green implies the
task is currently running.
If we had to have any failures, we would get a red box.
However, from what I can see here, all the tasks are a success.
If we had to press on one of the tasks themselves and go inside the logs, we would see what was actually
done at the code level.
Maybe we can have more details if we press on this one here and seeing what was actually done.
So here you can see some prints that we have.
These are for the video IDs.
Ultimately this Dag needs to produce a JSON file under the data directory.
So we will know that this Dag successfully completed by going back to VS code and seeing if the JSON
file was generated for today's date.
Let's look inside the data folder.
And as you can see we have a new JSON file.
Note we have briefly gone over airflow.
A final note that I want to emphasize for this lecture is that if we set the Dag to director on at 2
p.m., should the Docker containers be running at that time, then the Dag will be executed.
If, however, you have your laptop switched off, then the Docker environment won't be activated,
airflow won't be running, and as a result the DAGs will not be run at the scheduled time.
Now this contrasts greatly the setup that you would have in a production scenario.
In a production scenario, you would have a dedicated server where airflow is running 24 seven, be
it self-managed or on the cloud, most likely running on some Kubernetes environment.
But in our case, having a dedicated environment for airflow which incurs costs is out of scope for
this course.
Here we are building the toolset to then apply the knowledge on the workplace.
So that's all that I had to show you for this section.
Do feel free to go on the airflow UI and explore a bit more what we have just covered.
I'll see you in the next lecture to discuss the next Dag, which we will build for populating and updating
the data warehouse.