dbx-elt-utils 0.1.1

Librería de utilidades comunes para pipelines ELT orientados a Databricks (DLT).

DBX ELT Utilities

dbx-elt-utils es una librería de utilidades diseñada para simplificar y unificar la creación de pipelines Delta Live Tables (DLT) en Databricks. Proporciona métodos estandarizados para ingesta híbrida, setup automático del entorno (Local vs Nube) y limpieza de datos estructurados.

🚀 Instalación

Para uso en producción (Pipelines de Databricks):

pip install dbx-elt-utils

(Nota: La librería no fuerza la instalación interna de pyspark, previniendo conflictos con el Databricks Runtime).

Para desarrollo local (VS Code / Python Kernel): Incluye el SDK oficial para que tu PC pueda conectarse y simular ejecuciones.

pip install dbx-elt-utils[local]

---

🛠️ Cómo Inicializar un Notebook

Olvídate de configurar el espantoso sys.path. Solo necesitas estas 5 líneas al inicio de cualquier notebook (Bronze, Silver o Gold):

from dbx_elt_utils.notebook_utils import init_notebook

# Detecta automáticamente si estás en la Nube de Databricks o probando en local.
# Si estás en local, activa un "Mock" de DLT para que no tire error de importación.
notebook = init_notebook()

# Variables listas para usar:
env = notebook.env       # Ej: '_dev' o '_prod'
spark = notebook.spark   # Tu sesión de PySpark
dlt = notebook.dlt       # El módulo delta live tables (o el mock si estás local)

---

📦 Módulos Principales y Funcionalidad

Esta librería está dividida en 3 módulos lógicos:

1. ingest_utils.py (Ingesta Inteligente)

Diseñado para la capa Bronze. Su función principal es abstraer la complejidad técnica de AutoLoader y lectura de Unity Catalog.

• ingesta_hibrida(spark, origen, tipo="auto_detect", formato_archivo="json", env_suffix="_dev")
  • Qué hace: Detecta si tu origen es una ruta RAW (como abfss://...) o una tabla externa de Databricks para crear un Streaming DataFrame perfecto para DLT.
  • Súper poder: Si es una ruta (AutoLoader), activa mágicamente la opción schemaEvolutionMode="addNewColumns". Esto evita que el pipeline se rompa si mañana llega un CSV con más columnas de las esperadas.
  • Uso típico: df = ingesta_hibrida(spark, "abfss://[e-mail]/data.csv", formato_archivo="csv")

2. clean_utils.py (Limpieza y Estandarización)

Diseñado para la capa Silver.

• extraer_valor_array_string(columna)
  • Qué hace: Las APIs externas muchas veces entregan valores atrapados en arreglos de texto (ej. ["12345"]). Esta función recibe la columna de Spark, elimina corchetes y comillas limpiamente usando expresiones regulares, convirtiéndolo a texto puro (12345).
  • Súper poder: Si el string viene vacío ("[]"), lo convierte formalmente a un valor NULL real de base de datos.
  • Uso típico: df.withColumn("id_limpio", extraer_valor_array_string(col("id_crudo")))

3. notebook_utils.py (Pruebas Locales Seguras)

Ideal para tus ciclos de prueba en VS Code (Mock Local). Además del init_notebook(), incluye herramientas de protección:

• get_local_source_table(spark, source_official)

  • Qué hace: Cuando pruebas la capa Silver/Gold localmente, no quieres arriesgarte a sobrescribir tablas oficiales en Unity Catalog (table). Esta función averigua si corriste un "Test Bronze Local" antes. Si sí, fuerza a tu notebook Silver a leer los datos de prueba (temporary.table_tmp_sql). Si no, lee la oficial.

• clean_local_test_table(spark, source_table)

  • Qué hace: Aplica el patrón Read & Destroy. Destruye las tablas temporales (_tmp_sql) después de que haces df.show(), manteniendo el catálogo limpio.

• get_test_spark()

  • Qué hace: Crea e invoca la sesión de Databricks Connect.

• stop_local_spark()

  • Qué hace: El salvavidas principal. Al terminar una prueba local, mata el proceso persistente de Java/Databricks-Connect. Sin esta función, tu kernel de local de Python se quedaría colgado hasta que reinicies VS Code.
  • Uso típico: Poner siempre en la cláusula finally: de tus bloques de test.