bq_dq_plugin

Airflow plug-in that allows you to compare various metrics based on aggregated historical metrics and arbitrary thresholds.

Why?

Previously, Airflow’s BigQueryIntervalCheckOperator allowed you to compare the current day’s metrics with the metrics
of another day. This extends that functionality and allows you to compare it to aggregated metrics over an arbitrary window.
This is import for more robust data quality checks that are more resilient to noise and daily flucuation.

You could run the following query:

select count(*) as NumRecords, max(column1) as MaxColumn_1 
  from table
 where date_column = '{{ds}}'

which computes the number of rows and the max of column1 for today.
Now, compare it with the same metrics computed 7 days earlier:

select count(*), max(column1) 
  from {{table}}
 where date_column = timestamp_sub('{{ds}}', interval 7 day)

However, if you want more robust data quality checks you likely want to compare today’s metric
with an aggregated value (since that is less susceptible to daily flucutations/noise).

With this library, you can compare:

select count(*) as NumRecords, max(column1) as MaxColumn_1 
  from table
 where timestamp_trunc(date_column, Day) = timestamp_trunc('2019-01-01', Day)

which computes the number of records and the max of column1 for records that have date_column = '2019-01-01'.
Now, let’s compare the aggregated values:

with data as (
    select count(*) as NumRecords, max(column1) as MaxColumn_1, timestamp_trunc(date_column, Day) as Time
      from table 
     where date_column between '2018-01-01' and '2018-12-31' 
     group by Time
)
select avg(NumRecords), avg(MaxColumn_1)
  from data

Like the BigQueryIntervalCheckOperator, you can pass a dict which contains all metrics and their associated thresholds as key/value pairs.

In addition to this functionality, there will be some out-of-the-box checks including:

Numerical

num_records
percent_null
mean
std_dev
num_zero
min
median
max

Categorical

num_records
percent_null
num_unique
top
top_freq
avg_str_len

This plugin will allow you to create more complex custom checks as well.

Example usage (in a DAG)

from airflow import DAG
from datetime import datetime, timedelta
from bq_dq_plugin.operators.big_query_aggregate_check_operator import BigQueryAggregateCheckOperator
default_args = {
    'owner': 'Zachary Manesiotis', 
    'depends_on_past': False, 
    'start_date': datetime(2019, 12, 10), 
    'email': ['zacl.manesiotis@gmail.com'], 
    'email_on_failure': True, 
    'email_on_retry': False, 
    'retries': 3, 
    'retry_delay': timedelta(minutes=1)
} 
with DAG('Dag_ID', schedule_interval='@weekly', max_active_runs=15, catchup=False, default_args=default_args) as dag: 
    data_quality_check = BigQueryAggregateCheckOperator(
        task_id='data_quality_check', 
        table='`project.dataset.table`',
        metrics_thresholds={'count(*)': 1.5, 'max(Column1)': 1.6}, 
        date_filter_column='DateTime', 
        agg_time_period='Day', 
        start_date='2019-01-01', 
        end_date='2019-12-01', 
        gcp_conn_id=CONNECTION_ID, 
        use_legacy_sql=False, 
        dag=dag
    )
data_quality_check