Metric Calculator Engines Overview and Benchmarking

Checkita DataQuality contains various data quality metrics which all are computed during single pass over the data. Still, there are some metrics that are based on exact cardinality evaluation. Computation of this metrics requires either storing large state in memory (unique records) or shuffling data. First approach can cause OOM errors and the second one violates statement that data is read only once.

Checkita has to engines for metric computation: RDD-based engine and DF-based engine. Thus, RDD-based engine uses first approach to calculate cardinality-based metrics: metric calculators store in memory collection of all unique records. The DF-based engine, on the other hand, groups and shuffles the data to compute cardinality during latter reduce stage.

There are some additional points that need to be considered when choosing appropriate engine to run data quality job:

• RDD-based engine provide more flexibility when it is required to work closely with row data. Downside of such flexibility is that accessing row elements as Java-objects yields significant serialization/deserialization overhead.
• DF-based engine on the other hand operates using Spark UnsafeRow format which is stored off-heap and, in addition, has some performance leverage from using Tungsten whole-stage code generation. Problem here is that intermediate state of the calculator cannot be accessed and operated with.

Concluding the statements above, we recommend using DF-based engine for batch applications. While streaming applications support only RDD-based engine due to states of the calculators have to be merged between the processing windows and out of the Spark runtime.

Benchmarking

Benchmark tests below separated into two parts:

• evaluate all single-pass metrics (except cardinality-based ones)
• evaluate cardinality-based metrics

Along with running Checkita DQ jobs using either RDD-based of DF-based engine, the same metrics were computed using PySpark job in order to assess how well Checkita performs metric computation as compared to raw Spark code.

About Dataset

The New York Taxi dataset was chosen for processing as it contains massive amount of data collected since 2009.

For the purpose of DQ job run the data for 2009 year was selected from following sources:

https://d37ci6vzurychx.cloudfront.net/trip-data/yellow_tripdata_2009-01.parquet
https://d37ci6vzurychx.cloudfront.net/trip-data/yellow_tripdata_2009-02.parquet
https://d37ci6vzurychx.cloudfront.net/trip-data/yellow_tripdata_2009-03.parquet
https://d37ci6vzurychx.cloudfront.net/trip-data/yellow_tripdata_2009-04.parquet
https://d37ci6vzurychx.cloudfront.net/trip-data/yellow_tripdata_2009-05.parquet
https://d37ci6vzurychx.cloudfront.net/trip-data/yellow_tripdata_2009-06.parquet
https://d37ci6vzurychx.cloudfront.net/trip-data/yellow_tripdata_2009-07.parquet
https://d37ci6vzurychx.cloudfront.net/trip-data/yellow_tripdata_2009-08.parquet
https://d37ci6vzurychx.cloudfront.net/trip-data/yellow_tripdata_2009-09.parquet
https://d37ci6vzurychx.cloudfront.net/trip-data/yellow_tripdata_2009-10.parquet
https://d37ci6vzurychx.cloudfront.net/trip-data/yellow_tripdata_2009-11.parquet
https://d37ci6vzurychx.cloudfront.net/trip-data/yellow_tripdata_2009-12.parquet

Datasets are stored in Parquet format and are read as is during DQ job processing.

All parquet files has the same schema:

root
 |-- vendor_name: string (nullable = true)
 |-- Trip_Pickup_DateTime: string (nullable = true)
 |-- Trip_Dropoff_DateTime: string (nullable = true)
 |-- Passenger_Count: long (nullable = true)
 |-- Trip_Distance: double (nullable = true)
 |-- Start_Lon: double (nullable = true)
 |-- Start_Lat: double (nullable = true)
 |-- Rate_Code: double (nullable = true)
 |-- store_and_forward: double (nullable = true)
 |-- End_Lon: double (nullable = true)
 |-- End_Lat: double (nullable = true)
 |-- Payment_Type: string (nullable = true)
 |-- Fare_Amt: double (nullable = true)
 |-- surcharge: double (nullable = true)
 |-- mta_tax: double (nullable = true)
 |-- Tip_Amt: double (nullable = true)
 |-- Tolls_Amt: double (nullable = true)
 |-- Total_Amt: double (nullable = true)

Spark Configuration

All jobs in these test were run using the same Spark application configuration with following parameters:

"spark.driver.cores" = 2
"spark.driver.memory" = "4G"
"spark.executor.memory" = "12G"
"spark.executor.cores" = "4"
"spark.executor.instances" = "3"
"spark.shuffle.service.enabled" = "true"
"spark.sql.shuffle.partitions" = "12"
"spark.default.parallelism" = "12"
"spark.dynamicAllocation.enabled" = "false"
"spark.sql.orc.enabled" = "true"
"spark.sql.parquet.compression.codec" = "snappy"
"spark.sql.autoBroadcastJoinThreshold" = -1
"spark.sql.caseSensitive" = "false"
"spark.sql.legacy.timeParserPolicy" = "CORRECTED"

IMPORTANT For cardinality-based metric tests the parallelism has been increased to 1000:

"spark.sql.shuffle.partitions" = "1000"
"spark.default.parallelism" = "1000"

PySpark Job Run

First of all, we tried to compute the data quality metrics directly, using PySpark API and measured time, that job had taken.

Below is the code for computing single-pass metrics:

from pyspark.sql.types import StringType, IntegerType, DoubleType
from pyspark.sql.functions import col, lit, when, avg, sum as Fsum, min as Fmin, max as Fmax, datediff, \
    length, split, covar_pop, covar_samp, percentile_approx, levenshtein, stddev_pop, count, to_timestamp, upper

# spark = <initiate spark session>

ytd = spark.read.parquet('<path to folder with NY Taxi dataset>')

ytd.withColumn(
    "null_pmt_typ", when(col('payment_type').isNull(), lit(1)).otherwise(lit(0))
).withColumn(
    "empty_pmt_typ", when(col('payment_type') == lit(''), lit(1)).otherwise(lit(0))
).withColumn(
    "null_vendor", when(col('vendor_name').isNull(), lit(1)).otherwise(lit(0))
).withColumn(
    "date_1", to_timestamp(col('trip_pickup_datetime'), "yyyy-MM-dd HH-mm-ss")
).withColumn(
    "date_2", to_timestamp(col('trip_dropoff_datetime'), "yyyy-MM-dd HH-mm-ss")
).withColumn(
    "date_fmt_1", when(col('date_1').isNull(), lit(0)).otherwise(lit(1))
).withColumn(
    "date_fmt_2", when(col('date_2').isNull(), lit(0)).otherwise(lit(1))
).withColumn(
    "precision", length(col('fare_amt').cast(StringType())) - lit(1)
).withColumn(
    "scale", length(split(col('fare_amt').cast(StringType()), '\.').getItem(1))
).withColumn(
    "num_fmt_crit", (col('precision') <= lit(3)) & (col('scale') <= lit(2))
).withColumn(
    "num_to_cast", split(col('trip_pickup_datetime'), '-').getItem(0)
).select(
    count(lit(1)).alias('row_count'),
    Fsum(col('null_pmt_typ')).alias('null_vals'),
    Fsum(when(col('empty_pmt_typ') == lit(''), lit(1)).otherwise(lit(0))).alias('empty_vals'),
    (lit(1) - Fsum(col('null_pmt_typ') + col('null_vendor')) / lit(2) / count(lit(1))).alias('completeness'),
    Fmin(length(col('payment_type'))).alias('min_str'),
    Fmax(length(col('payment_type'))).alias('max_str'),
    avg(length(col('payment_type'))).alias('avg_str'),
    Fsum(when(length(col('payment_type').cast(StringType())) == lit(4), lit(1)).otherwise(lit(0))).alias('str_len'),
    Fsum(when(col('vendor_name').isin("VTS", "DDS", "CMT"), lit(1)).otherwise(lit(0))).alias('str_in_dmn'),
    Fsum(when(~col('payment_type').isin("Credit", "Cash"), lit(1)).otherwise(lit(0))).alias('str_out_dmn'),
    Fsum(when(col('payment_type') == lit("No Charge"), lit(1)).otherwise(lit(0))).alias('str_vals'),
    Fsum(when(col('tip_amt').rlike(r"^\d\d+\..+$"), lit(1)).otherwise(lit(0))).alias('rgx_match'),
    Fsum(when(~col('end_lat').rlike(r"^40\..*$"), lit(1)).otherwise(lit(0))).alias('rgx_mismatch'),
    Fsum(col('date_fmt_1') + col('date_fmt_2')).alias('fmt_date'),
    Fsum(when(col('num_fmt_crit'), lit(1)).otherwise(lit(0))).alias('fmt_number'),
    Fmin(col('total_amt')).alias('min_num'),
    Fmax(col('total_amt')).alias('max_num'),
    Fsum(col('total_amt')).alias('sum_num'),
    avg(col('total_amt')).alias('avg_num'),
    stddev_pop(col('total_amt')).alias('std_num'),
    Fsum(when(col('num_to_cast').cast(DoubleType()).isNull(), lit(0)).otherwise(lit(1))).alias('cast_num'),
    Fsum(when(col('passenger_count').isin(1, 2, 3), lit(1)).otherwise(lit(0))).alias('num_in_dmn'),
    Fsum(when(~col('passenger_count').isin(1, 2, 3, 4, 5), lit(1)).otherwise(lit(0))).alias('num_out_dmn'),
    Fsum(when(col('trip_distance') < lit(1.0), lit(1)).otherwise(lit(0))).alias('num_less_than'),
    Fsum(when(col('trip_distance') > lit(30.0), lit(1)).otherwise(lit(0))).alias('num_greater_than'),
    Fsum(when(
        (col('trip_distance') >= lit(5.0)) & (col('trip_distance') <= lit(10.0)), lit(1)
    ).otherwise(lit(0))).alias('num_btwn'),
    Fsum(when(
        (col('trip_distance') <= lit(1.0)) | (col('trip_distance') >= lit(10.0)), lit(1)
    ).otherwise(lit(0))).alias('num_not_btwn'),
    Fsum(when(col('tolls_amt') == lit(0.0), lit(1)).otherwise(lit(0))).alias('num_vals'),
    percentile_approx(col('fare_amt'), 0.5, 1000).alias('median_val'),
    percentile_approx(col('fare_amt'), 0.25, 1000).alias('first_quant_val'),
    percentile_approx(col('fare_amt'), 0.75, 1000).alias('third_quant_val'),
    percentile_approx(col('fare_amt'), 0.9, 1000).alias('get_quant_val'),
    Fsum(when(col('start_lon') == col('end_lon'), lit(1)).otherwise(lit(0))).alias('col_eq'),
    Fsum(when(datediff(col('date_1'), col('date_2')) < lit(1), lit(1)).otherwise(lit(0))).alias('date_dist'),
    Fsum(when(
        levenshtein(upper(col('start_lat').cast(StringType())), upper(col('end_lat').cast(StringType()))) < lit(3), lit(1)
    ).otherwise(lit(0))).alias('lvnst_dist'),
    (covar_pop(col('trip_distance'), col('fare_amt')) * count(lit(1))).alias('comoment'),
    covar_pop(col('trip_distance'), col('fare_amt')).alias('covariance'),
    covar_samp(col('trip_distance'), col('fare_amt')).alias('covarianceBessel')
).show(truncate=False, vertical=True)

Pyspark Single-Pass Metrics Test Results

Summary:

Longest tasks statistics (tasks that actually read all 12 dataset parquet files):

Pyspark Grouping Metrics Test Results

Distinct Values Metric

First test for cardinality-based metrics is to calculate number of distinct values. Below is the PySpark code to calculate number of distinct trip distances:

dist_vals = ytd.groupBy('trip_distance').agg(lit(1)).count()

print(dist_vals)

Summary:

Mapping tasks statistics (tasks that actually scanned all 12 dataset parquet files). Note that scanning parquet files for distinct column values is extremely fast!

Duplicate Values Metric

Let's calculate duplicated trips. These are trips for which following columns do match:

• vendor_name
• trip_pickup_datetime
• trip_dropoff_datetime
• start_lon
• start_lat
• end_lon
• end_lat

PySpark code to calculate number of duplicated trips is following:

dupl_vals = ytd.groupBy(
    'vendor_name', 
    'trip_pickup_datetime', 
    'trip_dropoff_datetime',
    'start_lon',
    'start_lat',
    'end_lon',
    'end_lat'
).agg(
    (count(lit(1)) - lit(1)).alias('cnt')
).select(Fsum(col('cnt'))).collect()[0][0]

print(dupl_vals)

Summary: Note that shuffling operations did spill lots of data to both memory and disk!

Mapping tasks statistics (tasks that actually scanned all 12 dataset parquet files).

Checkita RDD-Engine Job Run

Checkita RDD-Engine Single-Pass Metrics Test Results

Summary:

Longest tasks statistics (tasks that actually read all 12 dataset parquet files):

Checkita RDD-Engine Cardinality Metrics Test Results

Distinct Values Metric

Summary:

Longest tasks statistics (tasks that actually read all 12 dataset parquet files):

Duplicate Values Metric

Calculation failed with OOM error. The reason is that RDD-metric calculator for duplicate values must store all unique values within its state, i.e. within simple Java collection. For cases when number of unique records is high (in our case its about 170 million records) the executors will fail with OOM error.

Checkita DF-Engine Job Run

Checkita DF-Engine Single-Pass Metrics Test Results

Summary:

Longest tasks statistics (tasks that actually read all 12 dataset parquet files):

Checkita DF-Engine Cardinality Metrics Test Results

Distinct Values Metric

Summary:

Longest tasks statistics (tasks that actually read all 12 dataset parquet files):

Duplicate Values Metric

Summary: Note that shuffling operations did spill lots of data to both memory and disk!

Mapping tasks statistics (tasks that actually scanned all 12 dataset parquet files).

Overall Summary

Below are some general conclusions that can be made based on benchmark test results.

Single-Pass Metrics

The average input data for the task corresponds to a single parquet file and is following: * Average task input size is ~455 MiB. * Average task input records are 14241338.

Thus, DF-Engine provides roughly 5 times better performance for single-pass metrics as compared to RDD-Engine. This is due to RDD metric calculators require data serialization from DataFrame UnsafeRow format to JVM types while DF metric calculators run purely on Spark DF-API without any loss for data serialization.

Also, it could be noted Checkita DF-engine shuffled more data as compared to PySpark run. This is caused by the fact that DF metric calculators also collect row data for metric increment errors.

Cardinality-Based Metrics

Distinct Values Metric Calculation

Not that computation of distinct values via PySpark code written with use of Spark DF-API as well as via Checkita DF-Engine took credit for Parquet format specifics: the distinct values were obtained from metadata of the corresponding column chunks rather than from reading all data line by line. Checkita RDD-based engine on the other hand, read and processed all data records.

Thus, according to the test above, the Checkita DF-engine provided roughly 2 times better performance for distinct values calculation as compared to RDD-engine. In addition to that, DF-engine can take benefits of data columnar formats while RDD-engine will process all the records from dataset.

It also should be noted, that difference in performance between DF-Engine and RDD-engine will grow with the growth of column values cardinality. For the test above we have intentionally chosen column with low cardinality to prevent possible OOM errors for RDD-based distinct value metric calculator.

In the test above PySpark code run faster than Checkita DF-engine due to fact that we used the simplest way to calculate number of distinct values for the particular column while DF-metric calculator has more general and complex logic and, in addition, collects metric increment errors.

Distinct Values Metric Calculation

For calculation of duplicate values we have chosen tuple of columns with high cardinality (~170 million records). Thus, such RDD-based duplicate values metric calculator couldn't store such amount of unique records within its state and, therefore, application crushed with OOM error.

Note, that PySpark application performed better that Checkita DF-Engine for the same reasons as were given above for distinct values calculation.

Conclusions

From the test above it can be concluded that new Checkita DF-engine provides significantly better performance comparing to RDD-engine. We strongly recommend to use DF-engine for all batch applications.

However, DF metric calculators operate completely within Spark runtime. Currently, it is only possible to retrieve final metric result after data source is processed. RDD metric calculators on the other hand are more flexible and can be managed outside of Spark runtime. For that reason our streaming applications supports only RDD-engine for now as we need to merge calculators from different micro-batches outside the Spark runtime.

There are some plans to enhance DF metric calculators to make them support streaming application as well. :)

DQ Job Configuration files

Single-pass metrics

SRC = "ytd"
jobConfig: {
  jobId: "yellow_trip_data"
  jobDescription: "Data Quality job for NY Taxi Data"
  sources: {
    file: [
      {
        id: ${SRC},
        kind: "parquet",
        path: "/datalake/data/workspace/mlcib/ruadlu/yellow_trip_data"
        keyFields: ["vendor_name", "trip_pickup_datetime"]
      }
    ]
  }
  virtualSources: [
    {
      id: ${SRC}"_num_to_cast",
      kind: "select"
      parentSources: [${SRC}]
      expr: [
        "vendor_name",
        "trip_pickup_datetime"
        "split(trip_pickup_datetime, '-' )[0] as num_to_cast"
      ]
      keyFields: ["vendor_name", "trip_pickup_datetime"]
    }
  ]
  metrics: {
    regular: {
      rowCount: [{id: ${SRC}"_row_cnt", source: ${SRC}}]
      approximateDistinctValues: [{
        id: ${SRC}"_apprx_dist_vals", source: ${SRC},
        params: {accuracyError: 0.001}, columns: ["trip_distance"]
      }]
      nullValues: [{id: ${SRC}"_null_vals", source: ${SRC}, columns: ["payment_type"]}]
      emptyValues: [{id: ${SRC}"_empty_vals", source: ${SRC}, columns: ["payment_type"]}]
      completeness: [{id: ${SRC}"_completeness", source: ${SRC}, columns: ["vendor_name", "payment_type"]}]
      minString: [{id: ${SRC}"_min_str", source: ${SRC}, columns: ["payment_type"]}]
      maxString: [{id: ${SRC}"_max_str", source: ${SRC}, columns: ["payment_type"]}]
      avgString: [{id: ${SRC}"_avg_str", source: ${SRC}, columns: ["payment_type"]}]
      stringLength: [{
        id: ${SRC}"_str_len", source: ${SRC}, columns: ["payment_type"],
        params: {length: 4, compareRule: "eq"}
      }]
      stringInDomain: [{
        id: ${SRC}"_str_in_dmn", source: ${SRC}, columns: ["vendor_name"],
        params: {domain: ["VTS", "DDS", "CMT"]}
      }]
      stringOutDomain: [{
        id: ${SRC}"_str_out_dmn", source: ${SRC}, columns: ["payment_type"],
        params: {domain: ["Credit", "Cash"]}
      }]
      stringValues: [{
        id: ${SRC}"_str_vals", source: ${SRC}, columns: ["payment_type"],
        params: {compareValue: "No Charge"}
      }]
      regexMatch: [{
        id: ${SRC}"_rgx_match", source: ${SRC}, columns: ["tip_amt"],
        params: {regex: """^\d\d+\..+$"""}
      }]
      regexMismatch: [{
        id: ${SRC}"_rgx_mismatch", source: ${SRC}, columns: ["end_lat"],
        params: {regex: """^40\..*$"""}
      }]
      formattedDate: [{
        id: ${SRC}"_fmt_date", source: ${SRC},
        columns: ["trip_pickup_datetime", "trip_dropoff_datetime"],
        params: {dateFormat: "yyyy-MM-dd HH-mm-ss"}
      }]
      formattedNumber: [{
        id: ${SRC}"_fmt_number", source: ${SRC}, columns: ["fare_amt"],
        params: {precision: 3, scale: 2, compareRule: "inbound"}
      }]
      minNumber: [{id: ${SRC}"_min_num", source: ${SRC}, columns: ["total_amt"]}]
      maxNumber: [{id: ${SRC}"_max_num", source: ${SRC}, columns: ["total_amt"]}]
      sumNumber: [{id: ${SRC}"_sum_num", source: ${SRC}, columns: ["total_amt"]}]
      avgNumber: [{id: ${SRC}"_avg_num", source: ${SRC}, columns: ["total_amt"]}]
      stdNumber: [{id: ${SRC}"_std_num", source: ${SRC}, columns: ["total_amt"]}]
      castedNumber: [{id: ${SRC}"_cast_num", source: ${SRC}"_num_to_cast", columns: ["num_to_cast"]}]
      numberInDomain: [{
        id: ${SRC}"_num_in_dmn", source: ${SRC}, columns: ["passenger_count"],
        params: {domain: [1, 2, 3]}
      }]
      numberOutDomain: [{
        id: ${SRC}"_num_out_dmn", source: ${SRC}, columns: ["passenger_count"],
        params: {domain: [1, 2, 3, 4, 5]}
      }]
      numberLessThan: [{
        id: ${SRC}"_num_less_than", source: ${SRC}, columns: ["trip_distance"],
        params: {compareValue: 1.0, includeBound: false}
      }]
      numberGreaterThan: [{
        id: ${SRC}"_num_greater_than", source: ${SRC}, columns: ["trip_distance"],
        params: {compareValue: 30.0, includeBound: false}
      }]
      numberBetween: [{
        id: ${SRC}"_num_btwn", source: ${SRC}, columns: ["trip_distance"],
        params: {lowerCompareValue: 5.0, upperCompareValue: 10.0, includeBound: true}
      }]
      numberNotBetween: [{
        id: ${SRC}"_num_not_btwn", source: ${SRC}, columns: ["trip_distance"],
        params: {lowerCompareValue: 1.0, upperCompareValue: 30.0, includeBound: true}
      }]
      numberValues: [{
        id: ${SRC}"_num_vals", source: ${SRC}, columns: ["tolls_amt"],
        params: {compareValue: 0.0}
      }]
      medianValue: [{
        id: ${SRC}"_median_val", source: ${SRC}, columns: ["fare_amt"],
        params: {accuracyError: 0.001}
      }]
      firstQuantile: [{
        id: ${SRC}"_first_quant_val", source: ${SRC}, columns: ["fare_amt"],
        params: {accuracyError: 0.001}
      }]
      thirdQuantile: [{
        id: ${SRC}"_third_quant_val", source: ${SRC}, columns: ["fare_amt"],
        params: {accuracyError: 0.001}
      }]
      getQuantile: [{
        id: ${SRC}"_get_quant_val", source: ${SRC}, columns: ["fare_amt"],
        params: {accuracyError: 0.001, target: 0.9}
      }]
      getPercentile: [{
        id: ${SRC}"_get_percent_val", source: ${SRC}, columns: ["fare_amt"],
        params: {accuracyError: 0.001, target: 50.0}
      }]
      columnEq: [{id: ${SRC}"_col_eq", source: ${SRC}, columns: ["start_lon", "end_lon"]}]
      dayDistance: [{
        id: ${SRC}"_date_dist", source: ${SRC},
        columns: ["trip_pickup_datetime", "trip_dropoff_datetime"],
        params: {threshold: 1, dateFormat: "yyyy-MM-dd HH-mm-ss"}
      }]
      levenshteinDistance: [{
        id: ${SRC}"_lvnst_dist", source: ${SRC}, columns: ["start_lat", "end_lat"],
        params: {threshold: 3, normalize: false}
      }]
      coMoment: [{id: ${SRC}"_comoment", source: ${SRC}, columns: ["trip_distance", "fare_amt"]}]
      covariance: [{id: ${SRC}"_covariance", source: ${SRC}, columns: ["trip_distance", "fare_amt"]}]
      covarianceBessel: [{id: ${SRC}"_covariance_bessel", source: ${SRC}, columns: ["trip_distance", "fare_amt"]}]
    }
  }
  checks: {
    snapshot: {
      equalTo: [
        {id: ${SRC}"_row_cnt_chk", metric: ${SRC}"_row_cnt", threshold: 170896055}
      ]
    }
  }
}

Distinct values metric

SRC = "ytd"
jobConfig: {
  jobId: "yellow_trip_data"
  jobDescription: "Data Quality job for NY Taxi Data"
  sources: {
    file: [
      {
        id: ${SRC},
        kind: "parquet",
        path: "/datalake/data/workspace/mlcib/ruadlu/yellow_trip_data"
        keyFields: ["vendor_name", "trip_pickup_datetime"]
      }
    ]
  }
  metrics: {
    regular: {
      distinctValues: [{id: ${SRC}"_dist_vals", source: ${SRC}, columns: ["trip_distance"]}]
    }
  }
  checks: {
    snapshot: {
      equalTo: [
        {id: ${SRC}"_dist_vals_chk", metric: ${SRC}"_dist_vals", threshold: 6876}
      ]
    }
  }
}

Duplicate values metric

SRC = "ytd"
jobConfig: {
  jobId: "yellow_trip_data"
  jobDescription: "Data Quality job for NY Taxi Data"
  sources: {
    file: [
      {
        id: ${SRC},
        kind: "parquet",
        path: "/datalake/data/workspace/mlcib/ruadlu/yellow_trip_data"
        keyFields: ["vendor_name", "trip_pickup_datetime"]
      }
    ]
  }
  metrics: {
    regular: {
      duplicateValues: [{
        id: ${SRC}"_dupl_vals", source: ${SRC}, 
        columns: [
          "vendor_name", "trip_pickup_datetime", "trip_dropoff_datetime",
          "start_lon", "start_lat", "end_lon", "end_lat"
        ]
      }]
    }
  }
  checks: {
    snapshot: {
      equalTo: [
        {id: ${SRC}"_dupl_vals_chk", metric: ${SRC}"_dupl_vals", threshold: 34435}
      ]
    }
  }
}