We have covered most of what we need to know to perform the data quality tests, and now it's just a
matter of applying what we know inside airflow.
For this, we will use the airflow batch operator, which allows us to run batch commands in the batch
underscore command argument of the operator.
So let's first create a data quality subfolder under the DAGs folder.
And then here we define.
So the scripts and the script we will write all the base data quality logic for airflow Dag.
So we start off by first importing the logging module along with the batch operator.
We also set the logger variable.
And also the global variables for the parts where we will have the YAML files for the checks and configuration,
and the data source name which we defined in the configuration.
So I have the function ready.
So I will simply copy and paste and explain what we are doing.
So we define a function data quality.
And the argument that it takes is schema.
We define this as a parameter because it can either be the staging or the schemas.
And the try clause.
We are defining the batch operator by specifying the task id.
And here this is a function of the schema and also the batch command which is simply the.
So the scan command which we ran previously.
And finally we finish by returning the task as we have seen before.
If any exception occurs, we log it as an error and we raise it.
Now that we have this function set up, let's go inside the Main.py.
And what we can do is to simply copy the Dag that we have from the data warehouse section of this course,
and change it to use the logic for the data quality tests.
What we need to do is to again Change it again.
So this will become something like data quality.
The description will also change.
So you can say something like that.
So check the data quality on both layers in the database.
The schedule.
We need to set a time that runs after the previous tag.
So let's run it at 4:00 pm.
And the catch up we will keep it as false.
Finally, we set the tasks for both staging and core.
Here we also need to import the actual function.
So let's go to the very top.
And do the import.
And also we need to set the schema names for the staging and schemas.
We can also do this after the default arguments.
So let's set them here.
And now we have the variables for the staging and the schemas.
The final thing is to set the dependencies.
So the staging task will run before the core task.
So let's save.
And at this point we also have created the Dag for the data quality checks.
Let's go on the airflow UI and run this Dag.
Says the Dag manually we can simply unpause and let it run.
So we're seeing two dark greens here.
So it should signify that all was a success.
Let's check the logs.
As you can see for the staging schema we got the all is good.
No failures, no warning errors.
And for core, I'm expecting the same nodes.
And in fact it is.
So at this point we have now covered the data quality tests.
Feel free to add more tests in the YAML and explore more of solar's functionality.
In the next lecture we will go over the functional tests.
I will see you then.