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_excel

Carga el último archivo Excel desde /Volumes/bronze/excel/{env}/{source_file}/ y lo convierte a DataFrame Spark.

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

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