diff --git a/knowledge_base/failing_job_examples/src/schema_drift.py b/knowledge_base/failing_job_examples/src/schema_drift.py index 35053ab..463b3ab 100644 --- a/knowledge_base/failing_job_examples/src/schema_drift.py +++ b/knowledge_base/failing_job_examples/src/schema_drift.py @@ -1,18 +1,25 @@ # Databricks notebook source -from pyspark.sql.types import NumericType, StringType, StructField, StructType +from pyspark.sql.types import DecimalType, NumericType, StringType, StructField, StructType +# Simulated monetary amounts use a fixed-point decimal (not float) to avoid binary floating-point rounding drift. observed_source_schema = StructType( [ StructField("transaction_id", StringType(), False), - StructField("amount", StringType(), False), + StructField("amount", DecimalType(12, 2), False), ] ) -amount_type = observed_source_schema["amount"].dataType -if not isinstance(amount_type, NumericType): - raise TypeError( - "Schema contract violation in simulated source schema (no source table): " - f"expected amount to be numeric, found {amount_type.simpleString()}; " - f"observed schema {observed_source_schema.simpleString()}" - ) + +def validate_amount_schema(schema): + """Raise TypeError if the schema's amount field is not numeric.""" + amount_type = schema["amount"].dataType + if not isinstance(amount_type, NumericType): + raise TypeError( + "Schema contract violation in simulated source schema (no source table): " + f"expected amount to be numeric, found {amount_type.simpleString()}; " + f"observed schema {schema.simpleString()}" + ) + + +validate_amount_schema(observed_source_schema)