Documentación de `gss_bi_udfs`

set_spark

Configura una sesión Spark global para uso interno del módulo.

from pyspark.sql import SparkSession
from gss_bi_udfs import utils

spark = SparkSession.builder.getOrCreate()
utils.set_spark(spark)

get_spark

Obtiene Spark en este orden: parámetro explícito, sesión configurada por set_spark, builtins.spark (Databricks), o builder.getOrCreate().

get_env

Lee el entorno de ejecución desde la variable ENV.

get_env_catalog

Ajusta el catálogo según ambiente. Si ENV=pro devuelve el catálogo base; en otros casos agrega sufijo.

get_env_table_path

Compone el identificador final de tabla: catálogo ajustado + ruta de tabla.

get_schema_root_location

Consulta DESCRIBE SCHEMA EXTENDED y extrae RootLocation.

get_table_info

Resuelve información integral de tabla (nombre, path, existencia y metadatos de provider/type).

info = utils.get_table_info(full_table_name="fi_comunes.silver.dim_cliente")
print(info["full_table_name"], info["path"], info["exists"])

get_default_value_by_type

Devuelve un valor default de Spark (Column) según tipo de dato.

load_latest_parquet

Carga el parquet más reciente para la tabla solicitada desde /Volumes/bronze/{db}_{schema}/{env}/{table}/.

df_clientes = io.load_latest_parquet(
    spark=spark,
    data_base="timepro",
    schema="insudb",
    table="clientes",
    env="dev"
)

return_parquets_and_register_temp_views

Lee parquets según configuración y crea vistas temporales. Devuelve los DataFrames en un diccionario.

parquets_register_temp_views

Misma carga de parquets que la función anterior, pero sin devolver diccionario.

load_latest_delimited_file

Carga el último archivo delimitado desde la carpeta indicada. Está orientada a .csv, .txt y variantes similares leídas de forma nativa con Spark.

from gss_bi_udfs import io

df_clientes = io.load_latest_delimited_file(
    spark=spark,
    source_file="comercial/clientes_vigentes/periodo=2026-03",
    delimiter=";",
    header=True,
    encoding="latin1",
)

load_latest_excel_file

Carga el último archivo Excel .xls o .xlsx usando pandas y luego convierte el resultado a DataFrame Spark. Esta opción evita depender de com.crealytics.spark.excel en entornos donde el conector no está instalado.

from gss_bi_udfs import io

df_excel = io.load_latest_excel_file(
    spark=spark,
    source_file="comercial/clientes_vigentes/periodo=2026-03",
    header=True,
    sheet_name="Hoja1",
)

load_latest_file

Dispatcher genérico para cargar la última versión disponible de un archivo crudo. Soporta excel, xls, xlsx, csv, txt, tsv y delimited.

from gss_bi_udfs import io

df = io.load_latest_file(
    spark=spark,
    source_file="comercial/clientes_vigentes/periodo=2026-03",
    file_format="delimited",
    delimiter="|",
    header=False,
    infer_schema=False,
)

load_latest_excel

Función de compatibilidad para cargas Excel. Internamente delega a load_latest_excel_file con encabezado habilitado.

df_legacy = io.load_latest_excel(
    spark=spark,
    source_file="comercial/clientes_vigentes/periodo=2026-03",
    env="dev",
)

return_excels_and_register_temp_views

Carga Excels configurados, crea vistas temporales y devuelve diccionario de DataFrames.

excels_register_temp_views

Carga Excels y materializa vistas temporales sin retornar estructura adicional.

load_and_materialize_views

Dispatcher para acciones de carga: parquets/excels con o sin retorno.

tables_load = {
    "timepro": {"insudb": [{"table": "clientes", "view": "vw_clientes"}]}
}

out = io.load_and_materialize_views(
    action="return_parquets_and_register_temp_views",
    spark=spark,
    tables_load=tables_load,
    env="dev"
)

Comentarios Finales Sobre Archivos Crudos

Los archivos crudos versionados deben organizarse por dominio, dataset lógico y período de negocio. La convención esperada es que el proceso de ingesta publique archivos con timestamp ordenable en la última carpeta del path.

raw-files/
└── <dominio>/
    └── <nombre_logico_archivo>/
        └── periodo=<YYYY-MM>/
            └── <nombre_logico_archivo>_<YYYYMMDD_HHMMSS>.<ext>

Recomendaciones operativas:

save_table_to_delta

Guarda un DataFrame en formato Delta usando el modo de escritura indicado y lo registra en el metastore.

io.save_table_to_delta(
    spark=spark,
    df=df_clientes,
    catalog="fi_comunes",
    schema="silver",
    table_name="dim_cliente",
    mode="overwrite"
)

save_snapshot_to_delta

Guarda un snapshot de una tabla fact en Delta, borrando previamente el rango de snapshot/time_key que llega en el DataFrame.

io.save_snapshot_to_delta(
    spark=spark,
    df=df_fact_ventas,
    catalog="fi_comunes",
    schema="gold",
    table_name="fact_ventas_snapshot",
    snapshot_column="time_key"
)

add_hashid

Concatena columnas, calcula hash con xxhash64 y ubica la nueva PK técnica al inicio del DataFrame.

df_hash = transforms.add_hashid(
    df=df_clientes,
    columns=["id_cliente", "tipo_doc", "nro_doc"],
    new_col_name="sk_cliente"
)

get_default_record

Genera un único registro default usando el esquema de referencia.

df_default = transforms.get_default_record(spark, df_hash)
df_dim = df_hash.unionByName(df_default)

merge_scd2

Implementa Slowly Changing Dimension Tipo 2 sobre una tabla Delta, cerrando vigencias antiguas e insertando nuevas versiones.

merges.merge_scd2(
    spark=spark,
    df_dim_src=df_clientes.select("codigo_cliente", "segmento", "estado"),
    table_name="fi_comunes.silver.dim_cliente",
    business_keys="codigo_cliente",
    surrogate_key="sk_dim_cliente",
    eow_date="9999-12-31"
)