How to add Data sanity logic in airflow

user_13883723 0
Viewed 458

I am implemented airflow , can we add data sanity logic. suppose I have Task1 which do the following task

1.Read the data from the data source--RAW DATA.
2. do join with dimensional table to get the some relation detail product name etc.
3. Store output file some location after step 2.

There is a task2 that stored the output file into database. but before task2 execution i need to some data validation like count of RAW DATA should equal to the store output file count i.e after joining like count(raw_data) = count(raw_data_join_with_dimensional) , if it is true then trigger the Task2 else send the alert and failed the job.

airflow validation
2 Answers

For that use case a possible workflow could be:

check_op = SQLCheckOperator(
    task_id='check_task',
    sql='YOUR VALIDATION SQL',
    conn_id='YOUR CONN',
)

t2_op = YourNextOperator()

failure_op = EmailOperator(subject='check has failed', to='YOUR EMAIL', trigger_rule='one_failed')

check_op >> [t2_op, failure_op]

It works as follows:

  1. SQLCheckOperator runs the query against the DB. If query returns False the check has failed thus the operator will be in Failure state. If the query returns value the query consider as success thus the operator will be in Success state.
  2. EmailOperator will be triggered if SQLCheckOperator status is failure otherwise YourNextOperator will be triggered.
user_146244090

Edit: Go see @Elad's answer, there's a much more specific operator for this task.

The airflow.sensors.sql_sensor.SqlSensor can be used to build a task that can check data quality:

from airflow.sensors import sql_sensor


...

check_data_task = sql_sensor.SqlSensor(
  task_id="check_data",
  conn_id="YourConnectionIdentifier",
  sql="SELECT CASE WHEN data_is_valid THEN 1 ELSE 0 END ...",
  timeout=0
)

The key being that your sql argument return "at least one cell that contains a non-zero / empty string value" -- according to the documentation. The timeout=0 means that it will check once and fail if your data check query doesn't "pass"

user_1746520
Related