Pular para o conteúdo principal

Transformacao customizada no Databricks

Este exemplo mostra como manter o ContractForge como fronteira de contrato, enquanto um notebook Databricks executa um tratamento complexo entre varias tabelas.

O projeto completo esta em examples/real-world/databricks-movie-custom-transform.

Quando usar este padrao​

Use source.type: custom_transform quando uma transformacao fica mais clara em codigo revisado do que em YAML declarativo:

  • joins entre multiplas entradas;
  • engenharia de features;
  • metricas agrupadas ou com janelas;
  • bibliotecas externas;
  • regras de negocio que precisam de revisao de codigo.

O notebook e uma vinculacao nativa do Databricks. O contrato continua dono das entradas, saida esperada, destino, modo de escrita, qualidade, anotacoes, operacoes, evidencias e versionamento de deploy.

Fluxo de execucao​

contrato bronze_movie_ratings
contrato bronze_movie_titles
        |
contrato silver_movie_ratings
        |
notebook Databricks: prepare_movie_features.py
        |
contrato gold_movie_feature_summary custom_transform
        |
validacao, escrita, evidencia e versionamento pelo ContractForge

O Databricks Asset Bundle conecta o notebook como uma task nativa:

- 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"

A task gold do ContractForge depende dessa task de notebook.

Contrato gold​

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​

O exemplo usa pacotes PyPI no ambiente do job Databricks:

dependencies:
  - contractforge-core
  - contractforge-databricks

Execute com Databricks Asset Bundles:

databricks bundle validate
databricks bundle deploy
databricks bundle run movie_custom_transform

Se o workspace nao puder instalar a partir do PyPI, substitua as dependencias por caminhos de wheels publicados nos artefatos de release.