Introduction
We are building upon what we already have:
- Databricks series : Starting a new python project in Visual Studio Code (Part I)
- Databricks series : setting up a connection to Databricks from local IDE (Part II)
- Databricks series : Setting up a Databricks project in Visual Studio Code (part III)
- Databricks series : Deployment from Visual Studio Code (Part IV)
- Databricks series : Execute the job in Databricks (Part V)
- Databricks series : A configrable framework in Databricks (Part VI)
- Databricks series : loading strategies in Databricks (Part VII)
Great. We are now going to fully integrate Data Contracts into our existing Databricks project, so that it’s no longer just a standalone example, but an integral part of our data ingestion pipeline. The new approach: every dataset has its own Data Contract. Before data is written to Delta Lake, we verify that it complies with the contract.
Moreover, we are making the solution somewhat more robust than our previous version: we will first read the data in as strings, validate the values, and then convert them into the agreed-upon data types.
Change the structure of the project
databricks-transfer/ │ ├── databricks.yml │ ├── resources/ │ └── transfer_job.yml │ ├── src/ │ ├── __init__.py │ ├── main.py │ │ │ └── transfer/ │ ├── __init__.py │ ├── config_loader.py │ ├── contract_validator.py │ └── processor.py │ ├── config/ │ └── entities.json │ ├── contracts/ │ ├── customers.json │ └── transactions.json │ ├── data/ │ ├── customers.csv │ ├── customers_update.csv │ └── transactions.csv │ ├── requirements.txt └── README.md
We are including two contract files and one Python module for validation.
Create a datacontract for customers
{
"contract_name": "customers",
"version": "1.0",
"description": "Contract for customer master data",
"owner": "Data Engineering",
"allow_additional_columns": false,
"columns": [
{
"name": "customer_id",
"type": "string",
"nullable": false
},
{
"name": "first_name",
"type": "string",
"nullable": false
},
{
"name": "last_name",
"type": "string",
"nullable": false
},
{
"name": "country",
"type": "string",
"nullable": false
}
],
"primary_key": ["customer_id"],
"rules": {
"unique": ["customer_id"]
}
}
This contract contains the agreements with the producer of the customer data. A few key properties:
- version: allows us to manage changes to the contract.
- owner: the party responsible for the contract.
- allow_additional_columns: whether additional columns are permitted.
- primary_key: the column(s) used to identify a record.
- unique: which values must be unique.
We intentionally use a strict contract: if someone unexpectedly adds an extra column, the dataset will be rejected.
Connect the contract with entities.json
Lets open config/entities.json and replace the content with:
{ "customers": { "source": "/Volumes/main/demo/raw/customers.csv", "target": "main.demo.customers", "format": "csv", "mode": "overwrite", "contract": "contracts/customers.json" }, "customers_update": { "source": "/Volumes/main/demo/raw/customers_update.csv", "target": "main.demo.customers", "format": "csv", "mode": "merge", "merge_key": "customer_id", "contract": "contracts/customers.json" } }
For now, we are focusing on 'customers' and 'customers_update'. Both use the same contract, as they share the same data structure. We will add the 'transactions' contract once we have verified the exact columns in 'transactions.csv'. This ensures that we don’t define a contract that doesn’t match your actual file.
Build the contract validator
Now comes the most important part. Open the file src/transfer/contract_validator.py and replace the entire content with:
from pyspark.sql.functions import col, trim
TYPE_MAPPING = {
"string": "string",
"integer": "int",
"long": "bigint",
"double": "double",
"boolean": "boolean",
"date": "date",
"timestamp": "timestamp"
}
def validate_dataframe(df, contract):
errors = []
contract_name = contract["contract_name"]
version = contract["version"]
print(f"Validating contract: {contract_name}")
print(f"Contract version: {version}")
expected = {
item["name"]: item
for item in contract["columns"]
}
actual_columns = set(df.columns)
expected_columns = set(expected.keys())
# 1. Check schema
missing = expected_columns - actual_columns
unexpected = actual_columns - expected_columns
if missing:
errors.append(
f"Missing columns: {sorted(missing)}"
)
if unexpected and not contract.get(
"allow_additional_columns", False
):
errors.append(
f"Unexpected columns: {sorted(unexpected)}"
)
if errors:
raise ValueError(
"Schema validation failed:\n"
+ "\n".join(errors)
)
# 2. Check required fields and datatypes
validated_df = df
for name, definition in expected.items():
expected_type = definition["type"]
nullable = definition["nullable"]
if expected_type not in TYPE_MAPPING:
raise ValueError(
f"Unsupported datatype: {expected_type}"
)
source_col = col(name)
# Required fields cannot be NULL or empty
if not nullable:
invalid = df.filter(
source_col.isNull() |
(trim(source_col) == "")
).limit(1).count()
if invalid:
errors.append(
f"Column {name} cannot be NULL or empty"
)
# Validate and convert datatype
spark_type = TYPE_MAPPING[expected_type]
if expected_type != "string":
converted = source_col.cast(spark_type)
invalid = df.filter(
source_col.isNotNull() &
(trim(source_col) != "") &
converted.isNull()
).limit(1).count()
if invalid:
errors.append(
f"Invalid {expected_type} values "
f"in column {name}"
)
validated_df = validated_df.withColumn(
name, converted
)
# 3. Check primary key
for name in contract.get("primary_key", []):
if name not in expected:
errors.append(
f"Primary key column {name} "
"is not defined in contract"
)
# 4. Check unique constraints
unique_columns = set(
contract.get("rules", {}).get("unique", [])
)
unique_columns.update(
contract.get("primary_key", [])
)
for name in unique_columns:
if name not in expected:
errors.append(
f"Unique column {name} "
"is not defined in contract"
)
continue
duplicates = (
validated_df
.groupBy(name)
.count()
.filter(col("count") > 1)
.limit(1)
.count()
)
if duplicates:
errors.append(
f"Duplicate values found in {name}"
)
# 5. Final result
if errors:
raise ValueError(
"Data contract validation failed:\n"
+ "\n".join(errors)
)
print(
f"Contract {contract_name} "
f"version {version} PASSED"
)
return validated_df
What do we have now?
Validation Example
Schema Is `customer_id` missing?
Extra columns Is `phone_number` being added unexpectedly?
Null check Is `first_name` empty?
Data type Can an amount be converted to a `double`?
Primary key Is `customer_id` unique?
The validator returns an error message if a constraint is violated.
Integrate the validator in processor.py
The next step is to open: src/transfer/processor.py. Here, we ensure that validation always takes place before the loading strategy.
from delta.tables import DeltaTable
from transfer.contract_validator import validate_dataframe
def transfer_data(spark, config, contract=None):
source = config["source"]
target = config["target"]
file_format = config["format"]
mode = config["mode"]
print(f"Reading source: {source}")
print(f"Target table: {target}")
print(f"Loading mode: {mode}")
# Read everything as strings first
df = (
spark.read
.format(file_format)
.option("header", "true")
.option("inferSchema", "false")
.load(source)
)
# Validate before writing
if contract is None:
raise ValueError(
"No data contract configured for this entity"
)
df = validate_dataframe(df, contract)
record_count = df.count()
print(f"Validated records: {record_count}")
# Apply loading strategy
if mode == "overwrite":
(
df.write
.format("delta")
.mode("overwrite")
.saveAsTable(target)
)
elif mode == "append":
(
df.write
.format("delta")
.mode("append")
.saveAsTable(target)
)
elif mode == "merge":
merge_key = config["merge_key"]
if merge_key not in contract.get(
"primary_key", []
):
raise ValueError(
f"Merge key {merge_key} is not "
"defined as a contract primary key"
)
if spark.catalog.tableExists(target):
delta_table = DeltaTable.forName(
spark, target
)
(
delta_table.alias("target")
.merge(
df.alias("source"),
f"target.{merge_key} = source.{merge_key}"
)
.whenMatchedUpdateAll()
.whenNotMatchedInsertAll()
.execute()
)
else:
(
df.write
.format("delta")
.mode("overwrite")
.saveAsTable(target)
)
else:
raise ValueError(
f"Unsupported loading mode: {mode}"
)
print(
f"Successfully processed {record_count} records"
)
An important difference from our previous implementation: a MERGE can now only be performed if the merge key is also defined as the primary key in the Data Contract. Furthermore, duplicate keys in the source data are rejected before the MERGE is executed.
Modify main.py
Next step is to modify the main.py file.
import argparse
from pathlib import Path
from pyspark.sql import SparkSession
from transfer.config_loader import load_entity_config
from transfer.contract_validator import validate_dataframe
from transfer.processor import transfer_data
def main():
parser = argparse.ArgumentParser()
parser.add_argument(
"--entity",
required=True
)
args = parser.parse_args()
spark = SparkSession.builder.getOrCreate()
# Determine project root
project_root = Path(__file__).resolve().parents[1]
# Load entity configuration
config_path = project_root / "config" / "entities.json"
config = load_entity_config(
config_path,
args.entity
)
# Load associated data contract
contract_path = (
project_root / config["contract"]
)
import json
with open(contract_path, "r") as file:
contract = json.load(file)
# Execute transfer with contract
transfer_data(
spark,
config,
contract
)
if __name__ == "__main__":
main()
This completes the chain. The Job only passes the entity; the rest is automatically retrieved from the configuration files.
Adjust the DAB configuration
We need to ensure that the contract files are included in the deployment. Open: databricks.yml and check whether your configuration looks like this:
bundle:
name: databricks-transfer
include:
- resources/*.yml
sync:
include:
- src/**
- config/**
- contracts/**
- requirements.txt
targets:
dev:
default: true
mode: development
This synchronizes the JSON contracts with the rest of your project. The existing resources/transfer_job.yml file does not need to be modified. The job will continue to receive only the 'entity' parameter.
Testing: first, a successful validation
Now, we will check whether everything is working. Open the terminal in Visual Studio Code.
Validate the Bundle
Now we can deploy the project. Open the terminal in Visual Studio Code and navigate to the root of your project. Execute:
databricks bundle validate -t dev
Resulting in :
And if everything is right :
databricks bundle deploy -t dev
Next, you can start the job:
databricks bundle run transfer_job -t dev --params entity=customers
You can expect logging roughly like this:
Working directory: /Workspace/Users/xxxx@xxxx.com/.bundle/databricks-transfer/dev/files/src Project root: /Workspace/Users/xxxx@xxxx.com/.bundle/databricks-transfer/dev/files
Reading source: /Volumes/main/demo/raw/customers.csv Target table: main.demo.customers Loading mode: overwrite Validating contract: customers Contract version: 1.0 Contract customers version 1.0 PASSED Validated records: 4 Successfully processed 4 records
The exact logging may vary slightly, but contract validation must succeed before the table is updated.
databricks bundle run transfer_job -t dev --params entity=customers_update
This dataset must also comply with the same contract. Next, check:
SELECT * FROM main.demo.customers ORDER BY customer_id;
You are expecting five customers.
Testing an incorrect Data Contract
Now, we will intentionally introduce an error. This is important to demonstrate that our Data Contract integration is actually working. In Visual Studio Code, create: data/customers_invalid.csv
With the following content:
customer_id,first_name,last_name,country 1001,Jan,Jansen,Netherlands 1002,,Peeters,Belgium 1003,Anna,Smith,United Kingdom
Upload het bestand naar:/Volumes/main/demo/raw/customers_invalid.csv
Next, temporarily add this entity to config/entities.json:
"customers_invalid": { "source": "/Volumes/main/demo/raw/customers_invalid.csv", "target": "main.demo.customers", "format": "csv", "mode": "overwrite", "contract": "contracts/customers.json" }
Execute the folowing again :
databricks bundle deploy -t dev
And test the invalid schema :
databricks bundle run transfer_job -t dev --params entity=customers_invalid
Now the expected result is that the validation is failed
Task transfer_data FAILED:
ValueError: Data contract validation failed:
Column first_name cannot be NULL or empt
The job stops before the overwrite takes place. Consequently, your existing 'customers' table remains intact. This is a key advantage of this architecture: we validate before writing any data.
Now, this contract implementation is a custom solution. Next step is have a look at the Open Data Contract Standard (ODCS).
Hennie
Geen opmerkingen:
Een reactie posten