gss-bi-udfs 0.2.7

Utilidades reutilizables para Spark y Delta Lake en arquitecturas Lakehouse.

gss-bi-udfs

Creo modulo para guardar UDFs comunes a todas las areas de BI.

contrato de carga Excel

gss_bi_udfs.io.load_latest_excel_file carga el ultimo .xls o .xlsx versionado desde /Volumes/bronze/files/{env}/{source_file}/. La lectura del archivo se realiza con pandas y la creacion del DataFrame final se realiza con Spark, por lo que son dos etapas distintas de error y validacion.

Firma publica:

load_latest_excel_file(
    spark=None,
    source_file=None,
    env=None,
    header=True,
    sheet_name=0,
    schema=None,
)

schema es opcional y debe ser pyspark.sql.types.StructType. Cuando se informa, se aplica en spark.createDataFrame(pdf, schema=schema) para fijar el contrato de salida Spark. Es util cuando el Excel tiene columnas con formatos o tipos mixtos y Spark/Arrow no puede inferir un tipo unico.

El mismo parametro esta disponible en load_latest_file(..., schema=None) para cargas excel, xls y xlsx, y en load_latest_excel(..., schema=None) por compatibilidad. load_latest_excel ya no oculta fallas de lectura o conversion retornando None: propaga el error original con contexto. None queda reservado para el caso en que no se encuentre ningun archivo Excel.

Ejemplo con schema Spark:

from pyspark.sql.types import StructType, StructField, StringType, DoubleType
from gss_bi_udfs import io

schema = StructType([
    StructField("cuenta", StringType(), True),
    StructField("descripcion", StringType(), True),
    StructField("importe", DoubleType(), True),
])

df_excel = io.load_latest_excel_file(
    spark=spark,
    source_file="ospysa-muspeg/finanzas/parametria_tesoreria_mutual",
    env="dev",
    header=True,
    sheet_name=0,
    schema=schema,
)

configuracion de catalog/schema para gss_spark_flow

PipelineOrchestrator permite definir defaults por entorno:

export GSS_SPARK_FLOW_CATALOG=operaciones_dev
export GSS_SPARK_FLOW_SCHEMA=metadata

Con eso, PipelineOrchestrator(spark) aplica USE CATALOG y USE <schema> automaticamente. Si se pasan catalog/schema en el constructor, esos valores tienen prioridad sobre las variables de entorno.

configuracion de ejecucion de pipelines

La metadata incluye metadata.pipeline_execution_config para alojar inicialmente pipelines que se ejecutan en Fabric o Databricks:

campo	uso
pipeline_name	nombre logico del pipeline dentro del orquestador
medallion_layer	capa del medallero a la que pertenece el pipeline: bronze, silver o gold
execution_engine	fabric o databricks
fabric_group_id	workspace/group id de Fabric
fabric_pipeline_id	id del pipeline de Fabric
databricks_workspace	workspace de Databricks
databricks_path	path del job/notebook/pipeline en Databricks
parameters_json	parametros serializados como JSON
is_active	permite activar/desactivar sin borrar

Ejemplo:

orchestrator = PipelineOrchestrator(spark)

orchestrator.add_fabric_pipeline_config(
    pipeline_name="finanzas_fabric",
    group_id="fabric-group-id",
    pipeline_id="fabric-pipeline-id",
    medallion_layer="bronze",
    parameters={"periodo": "2026-04"},
)

orchestrator.add_databricks_pipeline_config(
    pipeline_name="finanzas_databricks",
    workspace="https://adb-xxx.azuredatabricks.net",
    path="/Shared/pipelines/finanzas",
    medallion_layer="silver",
    parameters={"periodo": "2026-04"},
)

configs = orchestrator.list_pipeline_execution_configs()

Con esa configuracion, cada step ya trae el pipeline_name. El orquestador usa ese nombre para buscar en metadata la configuracion de ejecucion del pipeline: motor, ids, workspace/path y parametros base.

medallion_layer se define a nivel de pipeline_name, no de dependencia. La regla esperada es que un mismo pipeline pertenezca siempre a una unica capa del medallero. Por ejemplo, 0.Master_Sancor_Load_Table puede quedar marcado como bronze, silver/dim_bup como silver y gold/dim_bup como gold.

plan = orchestrator.get_group_execution_plan(group_name)

for step in plan:
    decision = orchestrator.get_dataset_execution_decision(
        pipeline_name=step["pipeline_name"],
        dataset=step["dataset"],
    )

    if decision["should_process"]:
        result = orchestrator.execute_dataset(step)
        print(result)
    else:
        ctx = PipelineContext(
            spark,
            pipeline_name=step["pipeline_name"],
            task_name=step["dataset"],
            dataset=step["dataset"],
        )
        ctx.skip(
            reason=decision["reason"],
            decision=decision["decision"],
            reused_run_id=decision.get("reused_run_id"),
            reused_task_run_id=decision.get("reused_task_run_id"),
        )

execute_dataset dispara:

• Fabric: POST https://api.fabric.microsoft.com/v1/workspaces/{group_id}/items/{pipeline_id}/jobs/instances?jobType=Pipeline
• Databricks con job existente: POST {workspace}/api/2.1/jobs/run-now
• Databricks con notebook one-shot: POST {workspace}/api/2.1/jobs/runs/submit

Para Databricks, parameters_json debe incluir una de estas opciones de infraestructura:

• databricks_job_id o job_id
• existing_cluster_id
• new_cluster

Esos campos se usan para disparar la ejecucion y no se pasan como parametros al notebook. El resto de los parametros si se envian al notebook.

Tokens soportados por variable de entorno:

export GSS_SPARK_FLOW_FABRIC_TOKEN="..."
export GSS_SPARK_FLOW_DATABRICKS_TOKEN="..."

Tambien podes pasar parametros runtime por corrida:

orchestrator.execute_dataset(
    step,
    runtime_parameters={"periodo": "2026-04"},
)

Para validar el payload sin ejecutar:

orchestrator.execute_dataset(step, dry_run=True)

configuracion de ingesta bronze con Copy Data

La metadata separa la configuracion funcional del procesamiento (metadata.dataset_config) de la configuracion fisica de ingesta hacia bronze. Para cargas desde onpremise u origenes equivalentes, la configuracion del elemento Copy Data de Fabric vive en metadata.bronze_ingestion_config.

bronze_ingestion_config se define por dataset, no por pipeline_name + dataset: un dataset bronze debe tener una unica definicion canonica de ingesta. Si varios pipelines consumen ese dataset, deben depender del mismo dataset ya ingestado, no redefinir como se carga.

Para evitar repetir ids de Fabric en cada dataset, las conexiones se centralizan en metadata.connection_config:

campo	uso
connection_key	alias logico usado por otras tablas
fabric_connection_id	id de la conexion configurada en Fabric
connection_type	tipo de conexion, por ejemplo SQL Server o Azure Data Lake Storage Gen2
description	descripcion funcional
is_active	permite activar/desactivar sin borrar

Campos principales de metadata.bronze_ingestion_config:

campo	uso
dataset	dataset bronze ingestable
source_connection_key	conexion origen definida en connection_config
source_connection_type	tipo de origen esperado por Fabric; para onpremise, SQL Server
source_database	base de datos origen
source_schema	schema origen
source_table	tabla origen
source_query	consulta usada por Copy Data; puede ser select * o incluir filtro CDC
sink_connection_key	conexion destino definida en connection_config
sink_connection_type	tipo de destino esperado por Fabric; para bronze, Azure Data Lake Storage Gen2
sink_path_template	ruta parametrizada de destino
sink_file_format	formato de archivo destino; para bronze, Parquet
is_cdc	indica si la consulta aplica criterio CDC
is_active	permite activar/desactivar sin borrar

Ejemplo de ruta destino:

@concat(pipeline().parameters.DataBase, '/', pipeline().parameters.Schema, '/', pipeline().parameters.Table)

plan de ejecucion para Fabric como orquestador unico

Cuando Fabric coordina ejecuciones Fabric + Databricks, conviene generar un plan enriquecido una sola vez y pasarlo como JSON al Data Pipeline:

plan_json = orchestrator.get_group_execution_plan_json(group_name)

Ese JSON incluye, por cada step:

• pipeline_name y dataset
• execution_engine
• execution_config
• execution_parameters
• dependency_decision
• dataset_execution_decision

Ejemplo conceptual:

{
  "group_name": "modelo_estrella_finanzas_daily",
  "steps": [
    {
      "step": 1,
      "pipeline_name": "bronze/clientes",
      "dataset": "timepro.insudb.client",
      "execution_engine": "fabric",
      "can_execute": true,
      "should_process": true,
      "has_dataset_execution_policy": true
    },
    {
      "step": 2,
      "pipeline_name": "silver/clientes",
      "dataset": "fi_comunes.silver.clientes",
      "execution_engine": "databricks",
      "can_execute": true,
      "should_process": true,
      "has_dataset_execution_policy": false
    }
  ]
}

Para orquestar por medallero, el metodo recomendado es:

plan = orchestrator.get_group_medallion_execution_plan(group_name)

Ese plan devuelve, para cada target del grupo, tres etapas:

• bronze: lista de pipeline_name + dataset que puede dispararse en paralelo.
• silver: lista de pipeline_name + dataset en orden topologico.
• gold: lista de pipeline_name + dataset en orden topologico.

Ejemplo conceptual:

[
  {
    "step": 1,
    "pipeline_name": "gold/dim_bup",
    "dataset": "dim_bup",
    "steps": [
      {
        "step": 1,
        "layer": "bronze",
        "parallel": true,
        "items": [
          {"pipeline_name": "0.Master_Sancor_Load_Table", "dataset": "bup.bup.persons"},
          {"pipeline_name": "0.Master_Sancor_Load_Table", "dataset": "bup.bup.addresses"}
        ]
      },
      {
        "step": 2,
        "layer": "silver",
        "parallel": false,
        "items": [
          {"pipeline_name": "silver/dim_bup", "dataset": "dim_bup"}
        ]
      },
      {
        "step": 3,
        "layer": "gold",
        "parallel": false,
        "items": [
          {"pipeline_name": "gold/dim_bup", "dataset": "dim_bup"}
        ]
      }
    ]
  }
]

El plan trae solo los identificadores de ejecucion (pipeline_name y dataset) por item. Al momento de disparar cada item, se debe validar si puede ejecutarse con assert_can_execute(...) o inspeccionar get_dependency_diagnostics(...).

que representa cada tabla de metadata

tabla	representa
run	Ejecucion logica de un pipeline. Guarda inicio, fin, estado, disparador y datos operativos generales.
task_run	Detalle por tarea/dataset dentro de una ejecucion. Guarda estado, conteos, watermarks y decisiones de skip/reuso.
error_log	Errores asociados a una corrida o tarea.
watermark_state	Ultimo watermark confirmado por pipeline_name + dataset.
dataset_config	Configuracion funcional del procesamiento de un dataset: origen, destino, watermark y claves de merge.
connection_config	Catalogo centralizado de conexiones logicas usadas por configs tecnicas.
bronze_ingestion_config	Configuracion fisica de ingesta bronze por dataset fuente.
pipeline_dependency	DAG entre nodos ejecutables parent pipeline/dataset -> child pipeline/dataset.
pipeline_dependency_group	Targets que un grupo quiere materializar, normalmente nodos gold, con orden de prioridad.
pipeline_dependency_group_schedule	Calendarios asociados a grupos de ejecucion.
dataset_execution_policy	Politica de frescura propia de datasets fuente, principalmente bronze no-CDC.
pipeline_execution_config	Configuracion tecnica de ejecucion por pipeline_name y capa del medallero (medallion_layer).
lineage	Relacion input/output registrada por ejecucion.
data_quality	Resultados de controles de calidad.
schema_version	Versiones/migraciones aplicadas sobre la metadata.

Hay dos decisiones distintas:

decision	metadata	pregunta que responde
dependency_decision	metadata.pipeline_dependency	si el pipeline_name + dataset puede correr segun sus padres del DAG
dataset_execution_decision	metadata.dataset_execution_policy	si un dataset con politica configurada debe recargarse o puede reutilizar una carga fresca

dataset_execution_policy debe configurarse solamente para datasets cuya frescura se decide por politica propia, por ejemplo datasets onpremise cargados en Fabric. Para datasets silver/gold sin fila en esa tabla, el plan devuelve has_dataset_execution_policy = false y should_process = true; su ejecucion queda gobernada por pipeline_dependency.

politicas de recarga por dataset

La metadata incluye metadata.dataset_execution_policy para definir cada cuanto conviene volver a cargar un dataset fuente, independientemente de los grupos o pipelines que lo consuman. Esta tabla no define dependencias entre pipelines: esas dependencias viven en metadata.pipeline_dependency.

En la practica, dataset_execution_policy debe usarse para datasets onpremise o fuentes equivalentes cuya recarga se decide por frescura propia. Para datasets derivados como silver/gold, si no existe una fila en esta tabla, el orquestador no aplica politica de recarga y deja que pipeline_dependency gobierne si el step puede ejecutarse.

Para datasets CDC, no debe cargarse una fila en dataset_execution_policy cuando su actualizacion se gobierna por metadata.watermark_state y por dataset_config.watermark_column / dataset_config.watermark_type. En ese caso, la frecuencia efectiva de recarga del dataset queda determinada por el watermark y no por una politica de frescura independiente.

campo	uso
dataset	nombre del dataset al que aplica la politica
min_interval_minutes	intervalo minimo entre cargas exitosas
schedule_expr	expresion opcional de calendario del dataset
timezone	zona horaria asociada al calendario
skip_if_fresh	permite registrar SKIPPED si el dataset sigue fresco
is_active	permite activar/desactivar sin borrar

Ejemplo:

orchestrator = PipelineOrchestrator(spark)

orchestrator.add_dataset_execution_policy(
    dataset="timepro.insudb.intermedia",
    min_interval_minutes=1440,
    schedule_expr="0 0 6 * * ?",
    timezone="America/Argentina/Cordoba",
)

plan = orchestrator.get_group_execution_plan("modelo_estrella_finanzas_daily")

for step in plan:
    if not step["should_process"]:
        ctx = PipelineContext(
            spark,
            pipeline_name=step["pipeline_name"],
            task_name=step["dataset"],
            dataset=step["dataset"],
        )
        ctx.skip(
            reason=step["decision_reason"],
            decision=step["decision"],
            reused_run_id=step["reused_run_id"],
            reused_task_run_id=step["reused_task_run_id"],
        )

para compilar local

python3 -m build

para publicar local (manual)

python3 -m twine upload dist/*

publicar nueva version en pypi con github actions

Prerequisitos:

• Secret del repo configurado: PYPI_API_TOKEN
• Workflow: .github/workflows/python-publish.yml

Pasos:

1. Asegurate de tener los cambios listos en main.

   git checkout main
   git pull
   git status
   

2. Defini la nueva version semantica (X.Y.Z) y crea el tag vX.Y.Z.

   git tag -a v0.1.5 -m "Release v0.1.5"
   

3. Publica rama y tag en GitHub.

   git push origin main
   git push origin v0.1.5
   

4. Crea el release asociado al tag.
   • GitHub -> Releases -> Draft a new release
   • Seleccionar tag v0.1.5
   • Publicar (Publish release)
5. Verifica la corrida del workflow.
   • GitHub -> Actions -> Publish Python Package to PyPI
   • Debe finalizar en verde.
6. Verifica la version publicada en PyPI.
   • https://pypi.org/project/gss-bi-udfs/

Notas:

• El workflow valida que el tag tenga formato vX.Y.Z.
• La version del paquete se toma del tag (sin la v).
• Si falla el release, podes reintentar desde Actions con Run workflow y el input tag (ejemplo: v0.1.5).