Calidad de código

Un proyecto mantenible sitúa el paquete importable bajo `src/`, las pruebas bajo `tests/` y deja los notebooks como orquestadores finos. La lógica que transforma DataFrames vive en funciones con entradas y salidas explícitas; acceso a widgets, secretos, paths y escritura se encapsula en adaptadores. Así un cambio de notebook no es la única unidad desplegable ni verificable.

`pyproject.toml` declara el paquete, versión de Python, dependencias y herramientas. El layout `src/` evita importar accidentalmente el código desde el directorio de trabajo en vez del artefacto instalado. En Databricks Runtime 16.0 o superior, un notebook no debe usarse como módulo Python: refactoriza código compartido a archivos `.py` o a una wheel.

# src/orders/transforms.py
from pyspark.sql import DataFrame, functions as F

def valid_orders(df: DataFrame) -> DataFrame:
    return (
        df.where(F.col("order_id").isNotNull())
          .where(F.col("net_amount") >= 0)
          .withColumn("order_date", F.to_date("ordered_at"))
          .select("order_id", "customer_id", "order_date", "net_amount")
    )

Spark evalúa de forma perezosa, por lo que construir un DataFrame no demuestra que la transformación sea válida. Las pruebas deben materializar un resultado pequeño y comparar esquema, filas y casos borde. Crea fixtures mínimas que incluyan nulos, duplicados, timestamps y valores inválidos; evita copiar datasets productivos con información sensible.

Una transformación testable recibe DataFrames y parámetros ordinarios. Para comparar resultados, ordena por claves deterministas o utiliza utilidades de aserción de DataFrames disponibles en tu stack; nunca dependas del orden natural de particiones. Separa el contrato técnico —tipos y columnas— del contrato de negocio —por ejemplo, un único pedido por ID—.

# tests/test_transforms.py
from orders.transforms import valid_orders

def test_valid_orders_rejects_negative_amounts(spark):
    source = spark.createDataFrame(
        [(1, 7, "2026-07-20T10:00:00Z", 12.5),
         (2, 8, "2026-07-20T11:00:00Z", -4.0)],
        "order_id long, customer_id long, ordered_at string, net_amount double",
    )

    actual = valid_orders(source).orderBy("order_id").collect()

    assert [row.order_id for row in actual] == [1]
    assert str(actual[0].order_date) == "2026-07-20"

Una wheel empaqueta módulos y metadatos en un artefacto versionado. Declara dependencias directas en `pyproject.toml`, crea un lock o constraints para CI y fija versiones cuando la reproducibilidad lo exija. En jobs, instala la wheel como librería de la tarea; para compute clásico puedes almacenarla en workspace files o Unity Catalog Volumes según modo de acceso y runtime.

Serverless usa entornos y no admite init scripts. Fija paquetes y prueba el environment version seleccionado. Evita `%pip install` disperso por notebooks productivos: puede reiniciar Python, cambiar precedencia y hacer que dos tareas ejecuten código distinto. La promoción debe mover el mismo hash de wheel, no reconstruirla de fuentes diferentes en cada ambiente.

# pyproject.toml
[build-system]
requires = ["hatchling==1.27.0"]
build-backend = "hatchling.build"

[project]
name = "orders-pipeline"
version = "1.4.0"
requires-python = ">=3.11"
dependencies = [
  "pydantic==2.11.7",
]

[tool.pytest.ini_options]
testpaths = ["tests"]

Los tests unitarios verifican funciones y contratos con datos pequeños; no demuestran que el service principal pueda leer Unity Catalog, que la wheel se instale o que un `MERGE` sea idempotente. Las pruebas de integración despliegan en un catálogo aislado, ejecutan el entry point con identidad real y validan side effects. Deben usar nombres únicos por ejecución y limpiar sólo recursos etiquetados como temporales.

Databricks permite ejecutar pytest en el workspace y Databricks Connect puede acercar desarrollo local al compute remoto. Mantén la pirámide: muchas pruebas sin red, menos integraciones y pocas pruebas end-to-end. Marca suites y establece timeouts para que CI no ejecute por accidente una carga completa.

import pytest

@pytest.mark.integration
def test_merge_is_idempotent(spark, target_table, sample_orders):
    sample_orders.createOrReplaceTempView("incoming_orders")
    run_merge(spark, "incoming_orders", target_table)
    first = spark.table(target_table).count()

    run_merge(spark, "incoming_orders", target_table)
    second = spark.table(target_table).count()

    assert second == first

Una pipeline CI debe fallar pronto: lint y type check, tests unitarios, build de wheel, escaneo y validación del bundle. La integración se ejecuta con una identidad no humana y permisos mínimos en un target efímero o compartimentado. Conserva reporte de tests, hash del artefacto y commit para reconstruir la decisión de despliegue.

Los secretos pertenecen al proveedor de identidad o al gestor de secretos del CI; prefiere OAuth workload identity/service principal a PAT de usuario. La promoción a producción requiere que el mismo artefacto superado sea referenciado por la configuración prod. Añade rollback a la versión anterior y una comprobación post-deploy antes de dirigir triggers reales.

quality_gates:
  - name: unit
    command: pytest -m "not integration"
  - name: build
    command: python -m build
  - name: bundle_validate
    command: databricks bundle validate -t test
  - name: integration
    command: pytest -m integration
  - name: artifact_manifest
    command: sha256sum dist/*.whl