Skip to main content

Databricks custom transform

This example shows how to keep ContractForge as the contract boundary while a Databricks notebook performs complex treatment across multiple tables.

The complete project is in examples/real-world/databricks-movie-custom-transform.

Why use this pattern​

Use source.type: custom_transform when a transformation is better expressed as reviewed code than as contract YAML:

  • joins across multiple inputs;
  • feature engineering;
  • windowed or grouped metrics;
  • external libraries;
  • business logic that needs code review.

The notebook is a native Databricks binding. The contract still owns inputs, expected output, target, write mode, quality, annotations, operations, evidence and deployment versioning.

Runtime flow​

bronze_movie_ratings contract
bronze_movie_titles contract
        |
silver_movie_ratings contract
        |
Databricks notebook: prepare_movie_features.py
        |
gold_movie_feature_summary custom_transform contract
        |
ContractForge validation, write, evidence and deployment versioning

The Databricks Asset Bundle wires the notebook as a native task:

- task_key: prepare_movie_features
  notebook_task:
    notebook_path: ./notebooks/prepare_movie_features.py
    base_parameters:
      ratings_table: workspace.cf_movie_silver.s_movie_ratings
      movies_table: workspace.cf_movie_bronze.b_movie_titles
      output_table: workspace.cf_movie_tmp.movie_feature_engineering_output
      min_ratings: "10"

The gold ContractForge task depends on that notebook task.

Gold contract​

source:
  type: custom_transform
  intent: custom_treatment
  inputs:
    - alias: ratings
      table: workspace.cf_movie_silver.s_movie_ratings
    - alias: movies
      table: workspace.cf_movie_bronze.b_movie_titles

target:
  table: g_movie_feature_summary

layer: gold
mode: overwrite

transform:
  custom:
    name: movie_genre_feature_engineering
    output: workspace.cf_movie_tmp.movie_feature_engineering_output
    expected_columns:
      - genre
      - rating_band
      - rating_count
      - movie_count
      - user_count
      - avg_rating
      - rating_stddev
      - computed_at_utc
    parameters:
      min_ratings: 10

quality_rules:
  not_null: [genre, rating_band, rating_count, avg_rating]
  expressions:
    - name: positive_rating_count
      expression: rating_count > 0
      severity: abort
      message: Gold feature rows must represent at least one rating.
    - name: valid_average_rating
      expression: avg_rating BETWEEN 0 AND 5
      severity: abort
      message: Average rating must stay in the source rating scale.

extensions:
  databricks:
    custom_transform:
      notebook_path: ./notebooks/prepare_movie_features.py
      task_key: prepare_movie_features
      output_table: workspace.cf_movie_tmp.movie_feature_engineering_output
      base_parameters:
        ratings_table: workspace.cf_movie_silver.s_movie_ratings
        movies_table: workspace.cf_movie_bronze.b_movie_titles
        output_table: workspace.cf_movie_tmp.movie_feature_engineering_output
        min_ratings: "10"
    delta_properties:
      delta.enableChangeDataFeed: "true"

Deploy​

The example uses PyPI packages in the Databricks job environment:

dependencies:
  - contractforge-core
  - contractforge-databricks

Run it with Databricks Asset Bundles:

databricks bundle validate
databricks bundle deploy
databricks bundle run movie_custom_transform

If the workspace cannot install from PyPI, replace the dependencies with wheel paths published from the release artifacts.