[MUSIC PLAYING]
SPEAKER 1: Our next speaker is all the way from Singapore.
He is on the Hangout right now.
Chandu Kavar works as a data engineer at Grab Singapore.
Chandu has four plus years of experience working in Big Data
technologies such as Hive, Spark, Impala, Presto, Airflow,
EMR, et cetera.
Chandu enjoys playing table tennis and is also a blogger.
Chandu, can you--
CHANDU KAVAR: Yep.
SPEAKER 1: --enter the Hangout?
Yeah,
CHANDU KAVAR: OK.
Could you make me a presenter so I can--
SPEAKER 1: Yes.
CHANDU KAVAR: OK.
So my name is Chandu.
I am working for Grab Singapore.
So today, I'll talk about testing in Airflow
and what are the things that we can test.
So previously, I was working for ThoughtWorker,
so we are more focused on testing as well.
So this talk is a lightning talk,
so I will give you the high-level idea
about the testing.
So think about the testing in traditional web app
where we have a client server and database.
So we mostly tend to write unit tests, integration tests,
functional tests.
And if there is a very complex or a large scale application,
we used to write stress testing, or snapshot testing,
or a few other categories of test.
So what can we test in Airflow?
You might have seen there are some broken DAGs on the UI.
So when I joined Grab, on the [? broad ?] UI,
I can see a few DAGs.
Not everything would parse by Airflow.
But because of some typos or a cyclicity in the DAG--
or let's say you want to enforce all your developer to have
some mandatory parameters-- let's say owner of the DAG,
or you might have an email in the DAG.
So basically, you can enforce the mandatory parameter
as well.
Other thing you can test is how many tasks in that one DAG,
or the type of the task, task [INAUDIBLE] and operator.
So [INAUDIBLE] operator, or whatever the type of operator.
We can also test upstream and downstream dependencies
of each task.
We also might have a Custom Operator and sensor,
so we can test that logic of the Custom Operator and sensor.
There might be an integration test between two tasks,
if we have user xComs.
And at the end, we can test the end-to-end pipeline
to verify the result of the full pipeline.
We can run on the actual infrastructure, as well.
So if I will divide all these tests into categories,
so there will be five categories.
One is DAG validation test.
It will check the correctness of your DAG.
Basically, it will check the typos.
It will check the cyclicity of the DAG.
It will check the mandatory parameter on each DAG.
And this DAG validation test will
be common for all the DAGs.
You don't need to write tests for each DAG.
It will be high-level [? sanity ?] of all your DAGs.
The second will be a pipeline definition test.
It is more like snapshot testing in web app.
What it does is it just verifies whether your changes are
intentional or not.
Whenever you change the dependency of any task
or whether you introduce a new operator, this test will fail.
So you need to fix those tests, as well.
So this is like when you add all this,
you have to confirm with the test [INAUDIBLE]..
So the third one will be unit tests.
It will be for your custom operator and sensor.
So fourth and five, I'm working on it.
It is out of the scope for this presentation.
So let's take an example of a sample DAG.
The DAG we have is a Hello World DAG,
and it has three operators-- the DummyOperator, PythonOperator,
and one is custom operator, the MultiplyBy5Operator.
And then DummyOperator have a downstream dependency
on HelloOperator and MultiplyBy5Operator.
So what my custom operators do is, whatever argument you pass,
it will multiply by five, and it will return it.
So to write a DAG validation test,
the first test you can see here is test import DAGs.
What it does is it will check from the databags-- so
the DAG bag.
So in dagbag, as you might be aware,
it's all DAGs that Airflow has.
So it will verify the DAG will be passable to Airflow
or not in the unit test itself.
So you won't see broken DAGs on the UI, if you have this test.
So this test will check all the DAGs possible using Airflow.
The second test we have is a queue_present.
I'm checking that each and every DAG
should have a queue as a default argument.
If any DAG does not have a queue, this test will fail.
So this is how you can drive the DAG validation test.
You can add more tests based on your requirement.
So DAG definition test, as I mentioned,
it is a kind of snapshot.
What it does is it will check the number of tasks in the DAG.
So here, you can see I have a DAG ID, hello world.
And for this DAG, how many tasks in that particular DAG.
If you will add one more operator,
we have to change this test.
Then you can also check what are the type of tasks
using task ID.
Whatever task ID you have given, you can read it from dagbag.
And then you can check that this DAG should
contain this number of tasks.
You can also test the dependency as well.
So here is a test for dependency of dummy tasks.
So you can grab that task from the DAG bag,
and then you can have a SSN on the upstream and downstream
dependency.
So this is just to verify that whether you
are making the changes are intentional or not.
So the custom operator that we have
written, whatever the logic, is an execute method.
And we want to test the execute method,
so this is a test for that.
So what here I did is created a sample DAG
and then add the MultiplyBy5Operator in the DAG.
And then I had created a task instance.
If I will execute this task instance
using the value as a 10, then it should return me a 50.
So this is how you can add all your tests
to verify your operator logic or custom sensor logic.
So all the codes related to this presentation and a few more,
you can find in this report, for more reference.
And also, I have a blog post on medium for first three.
And I'm working on fourth and five.
Yeah.
That's it.
[APPLAUSE]
SPEAKER 1: Questions for Chandu, anyone?
CHANDU KAVAR: OK, thanks.
SPEAKER 1: There is one question.
Hold on.
CHANDU KAVAR: Oh, OK.
AUDIENCE: So a quick question.
So I'm curious that you have mentioned about operators
that can safely test locally.
What if you have operators that you want on some remote system,
say, a Spark job operator.
How would you test in that case?
CHANDU KAVAR: So I think that whatever
the operator that provided by Airflow that I think
already tested in Airflow code base.
Whatever the custom operator that you will write,
you need to test only that, all right?
And all the integration between your pipeline, you
will have an end-to-end pipeline test on small scale.
Basically, you will set up some data,
then you will run whole pipeline.
And you will [INAUDIBLE] on all your output
at the end of your DAG.
AUDIENCE: Got you.
Thanks.
Is this something you are already doing,
or that's something you plan to do?
CHANDU KAVAR: So I am working on the integration test
and end-to-end pipeline test.
Might be using Docker or some [INAUDIBLE]..
So it will have the same setup, like a product [? sim. ?]
So you can test all your pipeline
based on the [INAUDIBLE].
If someone will change that particular DAG,
that end-to-end pipeline [INAUDIBLE]..
It will run on sample data, and it
will verify the output of that particular pipeline.
AUDIENCE: Cool.
Thanks.
CHANDU KAVAR: Yeah.
Welcome.
SPEAKER 1: Thank you, Chandu.
CHANDU KAVAR: You're welcome.
[APPLAUSE]
[MUSIC PLAYING]