Introduction

Why Is Data Quality Testing Important?

• Data loss: Missing records or incomplete migration.

• Schema mismatches: Differences in data formats between on-prem and cloud.

• Inconsistent records: Discrepancies between the source (on-prem) and target (Databricks) datasets.

• Inaccurate data transformations: Errors caused by transformations applied in Databricks that don't mirror the on-prem environment.

Key Data Quality Dimensions to Consider

• Completeness: All required data fields and records must be present.

• Accuracy: Data must reflect correct values and be free from errors.

• Consistency: Data should remain consistent across different systems and formats.

• Validity: The data must conform to specified formats and ranges.

• Uniqueness: There should be no duplicate records.

Preparing Data for Testing

Choosing Between Full and Sample Datasets

Best Practices for Sampling

• Random Sampling: Random sampling is a method where a random subset of the data is selected from the overall dataset. This method provides a quick and unbiased way to test the pipeline and validate data without having to process the entire dataset. It's particularly useful for general-purpose testing during the early stages of migration or development, where broad coverage is required but performance constraints are a concern.

-- Random sample of 1% data
SELECT * FROM source_table WHERE RAND() <= 0.01;

• Stratified Sampling: Stratified sampling involves dividing the data into distinct categories or "strata" and sampling from each category proportionally. This method ensures that all relevant categories of data are represented in the sample. For example, in telco data, categories may include customer regions, customer types (e.g., prepaid vs. postpaid), or types of usage (e.g., mobile data, voice, SMS).
Example(Stratified sampling by customer_type):

-- Stratified sample, selecting 1% from each customer type 
SELECT * FROM billing_data WHERE customer_type = 'prepaid' 
AND RAND() <= 0.01 UNION ALL SELECT * FROM billing_data WHERE 
customer_type = 'postpaid' AND RAND() <= 0.01;

• Edge Case Sampling: Edge case sampling focuses on manually selecting records that are known to cause issues or fall outside the normal distribution of the data. For example, in telco data, this might include customers with extraordinarily high data usage or records with missing or inconsistent values. This method is used for testing data quality or handling anomalies
Example (Edge case sampling for high download data usage):

-- Select records where data usage exceeds 100 GB
SELECT * FROM usage_data
WHERE dl_usage_gb > 100;

Data Required to Test Pipelines

• On-prem Input: The data produced by on-prem systems before any transformation.

• On-prem Output: The final output produced by the on-prem pipeline after transformation.

• Databricks Input: The data ingested by Databricks pipelines (usually the same as on-prem input but after ingestion into the cloud).

• Databricks Output: The final output produced by the Databricks pipeline.

Setting Up Pytest for Spark and DataFrames

To enable automated testing, we will use Pytest along with Spark to compare the datasets. Unit Test Best Practices with Pytest

• Modular Tests: Each test case should be independent and reusable.

• Fixtures: Use Pytest fixtures to set up and reuse common resources like Spark sessions and configurations.

• Assertions: Provide clear and informative assertion messages.

• Test Structure: Follow the Arrange-Act-Assert pattern for clarity.

Configuration for Data Quality Parameters

{
  "application_name": "DataValidationApp",
  "pipeline_configuration": {
    "dbx_output_configuration": {
      "target_table_name": "your_delta_table"
    },
    "onprem_configuration": {
      "format": "parquet",
      "source_data_s3_path": "s3://your-bucket/onpremise-source-data",
      "output_data_s3_path": "s3://your-bucket/onpremise-output-data"
    },
    "data_quality_configuration": { 
       "null_checks": ["column1", "column2"], 
       "range_checks": { 
        "arpu": {"min": 100, "max": 5000}, 
        "call_duration": {"min": 0, "max": 60} }
       },
       "unique_checks": ["imei", "phone"],
       "statistical_checks": ['arpu', 'data_usage'],
       "aggregation_checks": {"call_duration": "sum", "arpu": "avg"},
       "min_max_checks": ["call_duration", "arpu"],
       "aggregation_checks": { 
          "sum(call_duration)": "sum", 
          "avg(arpu)": "avg" 
        },
         "date_format_checks": ["date_col1", "date_col2"]
  }
}

Spark Session Fixture

@pytest.fixture(scope="module")
def spark():
    config_path = "./config.json"
    spark, pipeline_config = initialize_pipeline(config_path)
    yield spark, pipeline_config

DataFrame Fixture

@pytest.fixture(scope="module")
def dataframes(spark):
    spark_session, pipeline_config = spark
    onprem_config = pipeline_config.onprem_configuration
    endpoint_config = pipeline_config.endpoint_configuration

    # Load on-prem data from S3
    onprem_df = (
        spark_session.read.format(onprem_config.format)
        .option("recursiveFileLookup", "true")
        .option("inferSchema", "true")
        .load(onprem_config.output_s3_path)
    )

    # Load Databricks data from the Delta table
    delta_df = spark_session.read.table(endpoint_config.target_table_name)

    return onprem_df, delta_df

Data Quality Testing Approaches

Completeness Tests

• Row Count Matching

def test_row_count_matching(dataframes):
    onprem_df, delta_df = dataframes
    assert onprem_df.count() == delta_df.count(), "Row count mismatch between on-premise and Databricks output"

• Null Count Validation

def test_null_counts(dataframes):
    onprem_df, delta_df = dataframes
    onprem_null_counts = {col: onprem_df.filter(onprem_df[col].isNull()).count() for col in onprem_df.columns}
    delta_null_counts = {col: delta_df.filter(delta_df[col].isNull()).count() for col in delta_df.columns}

    assert onprem_null_counts == delta_null_counts, f"Null count mismatch: {onprem_null_counts} != {delta_null_counts}"

Consistency Tests

• Data Consistency

def test_data_consistency(dataframes):
    onprem_df, delta_df = dataframes
    diff_df = onprem_df.subtract(delta_df).union(delta_df.subtract(onprem_df))
    assert diff_df.count() == 0, "Data inconsistency between on-premise and Databricks output"

• Schema Validation

def test_schema_validation(dataframes):
    onprem_df, delta_df = dataframes
    onprem_schema = onprem_df.dtypes
    delta_schema = delta_df.dtypes
    assert onprem_schema == delta_schema, f"Schema mismatch: {onprem_schema} != {delta_schema}"

Uniqueness Tests

• Unique Values Check

def test_unique_values(dataframes):
    onprem_df, delta_df = dataframes
    key_columns = pipeline_configuration.["data_quality_configuration"]["unique_checks"]
    for column in key_columns:
        onprem_unique = onprem_df.select(column).distinct().count()
        delta_unique = delta_df.select(column).distinct().count()
        assert onprem_unique == delta_unique, f"Unique value mismatch for column {column}"

• Duplicate Records

def test_duplicate_records(dataframes):
		onprem_df, delta_df = dataframes
		key_columns = pipeline_configuration["data_quality_configuration"]["unique_checks"]
		onprem_duplicates = onprem_df.groupBy(key_columns).count().filter("count > 1").count()
		delta_duplicates = delta_df.groupBy(key_columns).count().filter("count > 1").count()
		assert onprem_duplicates == delta_duplicates, "Duplicate record count mismatch"

Validity Tests

• Date Format Consistency

def test_date_format_consistency(dataframes):
	date_columns = ["date_col"]
	onprem_df, delta_df = dataframes
	for column in date_columns:
		onprem_date_formats = onprem_df.select(column).distinct().collect()
		delta_date_formats = delta_df.select(column).distinct().collect()
		assert onprem_date_formats == delta_date_formats, f"Date format consistency mismatch for column {column}"

• Column Value Range

def test_column_value_range(dataframes, config):
		onprem_df, delta_df = dataframes
		for column, bounds in pipeline_configuration.["data_quality_configuration"]["range_checks"].items():
				min_val, max_val = bounds["min"], bounds["max"]
				onprem_range_valid = onprem_df.filter((onprem_df[column] < min_val) | (onprem_df[column] > max_val)).count() == 0
				delta_range_valid = delta_df.filter((delta_df[column] < min_val) | (delta_df[column] > max_val)).count() == 0
				assert onprem_range_valid and delta_range_valid, f"Column value range mismatch for column {column}"

Accuracy Tests

• Aggregate Validation

def test_compare_aggregated_values(dataframes):
		onprem_df, delta_df = dataframes
		aggregate_checks = pipeline_configuration.["data_quality_configuration"]["aggragate_checks"]
		# Build the aggregation dictionary for Spark
		aggregations = {key.split('(')[1].replace(')', ''): value for key, value in aggregation_checks.items()}
		# Perform the aggregation on both the on-prem and delta dataframes delta_agg = delta_df.agg(aggregations).collect()[0]
		onprem_agg = onprem_df.agg(aggregations).collect()[0]
		# Validate each aggregation from the config
		for agg_key, agg_function in aggregation_checks.items():
				column_name = agg_key.split('(')[1].replace(')', '')
				assert delta_agg[f"{agg_function}({column_name})"] == onprem_agg[f"{agg_function}({column_name})"], \ f"{agg_function.capitalize()} for {column_name} mismatch between on-premise and Databricks output"

• Min and Max Value Validation

def test_min_max_values(dataframes):
		onprem_df, delta_df = dataframes
		columns = pipeline_configuration.["data_quality_configuration"]["min_max_checks"]
		for column in columns:
				onprem_min = onprem_df.agg({column: "min"}).collect()[0][0]
				delta_min = delta_df.agg({column: "min"}).collect()[0][0]
				onprem_max = onprem_df.agg({column: "max"}).collect()[0][0]
				delta_max = delta_df.agg({column: "max"}).collect()[0][0]
				assert onprem_min == delta_min, f"Min value mismatch for {column}"
				assert onprem_max == delta_max, f"Max value mismatch for {column}"

• Statistical Summary

def test_statistical_summary(dataframes):
		numerical_columns = pipeline_configuration.["data_quality_configuration"]["statistical_checks"]
		onprem_df, delta_df = dataframes
		for col in numerical_columns:
				onprem_summary = onprem_df.describe([col]).collect()
				delta_summary = delta_df.describe([col]).collect()
				assert onprem_summary == delta_summary, f"Statistical summary mismatch in column {col}"

Conclusion