Replica de Catalogos Maestros de Databricks a GCP

El proceso de réplica desde Databricks hacia Google Cloud Platform (GCP) es completamente dinámico y está guiado por metadatos. El comportamiento del Job, el enrutamiento de los datos y las credenciales de destino no están programados de forma rígida (hardcoded), sino que se gobiernan mediante dos tablas de control alojadas en el esquema de configuración:

1. cat_master_aplicaciones.catalogos.catalogos: Define qué tablas/vistas se replican y a qué ambientes aplican.
2. cat_master_aplicaciones.catalogos.conexiones: Almacena las cadenas de conexión, hosts, credenciales y metadatos de infraestructura de los destinos de base de datos en GCP.

Mecanismo de Vinculación y Mapeo Dinámico

El Proceso de Réplica de Databricks opera de manera inteligente realizando un cruce (match) entre ambas tablas utilizando el nombre del ambiente como clave relacional.

---

1. Estructura y Diccionario

1.1. Tabla catalogos

Esta tabla funciona como el inventario y registro de alta del sistema. Controla qué objetos entran al flujo del pipeline, cuáles son sus llaves de negocio y determina de forma lógica hacia qué ambientes se deben enrutar los datos.

Campo	Tipo de Dato	¿Permite Nulos?	Descripción / Regla de Negocio
catalog_id	BIGINT	No (Identity)	Identificador único autoincremental que se genera automáticamente al dar de alta un catálogo.
catalog_name	STRING	No	Nombre técnico exacto de la tabla o vista física en el origen (ej: alp_cat_ar_clientes_v).
description	STRING	Sí	Nombre común, descripción funcional o propósito de negocio del catálogo.
primary_key	STRING	Sí	Campo o campos que fungen como Llave Primaria en el destino. Si es compuesta, se separan por comas.
database_name	STRING	No	Nombre de la base de datos origen dentro de Databricks (ej: cat_master_aplicaciones).
schema_structure	STRING	No	Máscara o comodín (ej: % o sch_%_layer) que usa el Job para resolver el esquema dinámicamente por entorno (donde % = ENTORNO lúdico de ejecución: bronze, silver o gold).
appl	STRING	Sí	Nombre o etiqueta de la aplicación global consumidora (ej: Catalogo Maestro, Alpura One).
bronze	BOOLEAN	Sí	Flag de Envío (DEV): Si es true, el Job procesará este catálogo hacia el esquema de DEV en GCP.
silver	BOOLEAN	Sí	Flag de Envío (UAT): Si es true, el Job procesará este catálogo hacia el esquema de UAT en GCP.
gold	BOOLEAN	Sí	Flag de Envío (PROD): Si es true, el Job procesará este catálogo hacia el esquema PROD en GCP.
dn_activo	BOOLEAN	No (Default true)	Indicador de borrado lógico (Soft Delete). Si está en false, el Job ignora el catálogo por completo.
dd_fecha_creacion	TIMESTAMP	No (Default Now)	Registro de auditoría: Fecha y hora exacta en la que se dio de alta el catálogo.
dd_usuario_creador	STRING	Sí	Correo electrónico o ID del usuario técnico que realizó la inserción en la tabla de control.
dd_fecha_modificacion	TIMESTAMP	Sí	Fecha y hora de la última actualización realizada sobre los parámetros de este catálogo.
dd_usuario_modificador	STRING	Sí	Correo electrónico o ID del último usuario que editó los metadatos de configuración.

---

1.2. Tabla conexiones

Esta tabla almacena de forma centralizada los parámetros de infraestructura, cadenas de red y credenciales de acceso para las bases de datos PostgreSQL (Cloud SQL) hospedadas en Google Cloud. Su objetivo primordial es el aislamiento seguro de los accesos.

Diccionario de Columnas

Campo	Tipo de Dato	¿Permite Nulos?	Descripción / Regla de Negocio
id_conexion	BIGINT	No (Identity)	Identificador único autoincremental de la configuración de red.
conexion_name	STRING	No	Clave de Enlace: Debe llamarse exactamente 'bronze', 'silver' o 'gold' para empatar con los flags de la tabla de catálogos.
appl	STRING	Sí	Sistema asociado que consume el pipeline de datos (ej: Catalogo Maestro).
instancia	STRING	Sí	Nombre identificador del recurso físico asignado dentro de Google Cloud SQL (ej: algcpcloudsqldev).
project_id	STRING	Sí	El ID del proyecto dentro de la consola de GCP (ej: alpura-hub-dev, alpura-hub-uat).
project_number	STRING	Sí	Número interno de registro del proyecto asignado por Google Cloud.
location	STRING	Sí	Región geográfica y zona de disponibilidad donde vive el servidor de base de datos (ej: us-central1-a).
tier	STRING	Sí	Especificación de hardware o tamaño de la máquina virtual de base de datos (ej: db-perf-optimized-N-8).
name	STRING	Sí	Nombre descriptivo de la instancia de base de datos en GCP.
version	STRING	Sí	Motor de base de datos y versión instalada (ej: POSTGRES_17).
driver_class	STRING	Sí	Clase del controlador JDBC requerida por Databricks para inicializar la conexión (ej: org.postgresql.Driver).
host	STRING	No	Dirección IP (generalmente de la red privada VPC) del servidor PostgreSQL de destino.
port	INT	No	Puerto de red para la escucha de peticiones a la base de datos (por defecto para Postgres: 5432).
database_name	STRING	No	Nombre de la base de datos destino donde se persistirá la información (ej: omnicanal).
schema_name	STRING	No	Esquema específico dentro de la base de datos que contendrá las tablas replicadas (ej: general_catalogs).
jdbc	STRING	Sí	Cadena completa de conexión JDBC formateada (opcional si el Job la construye dinámicamente).
uid	STRING	No	Nombre del usuario/rol de base de datos de GCP con privilegios de escritura (ej: msa_app_arq_catalog).
pwd	STRING	No	Contraseña de acceso asociada al usuario. Almacenada bajo políticas de cifrado/ofuscación.
dn_activo	BOOLEAN	No (Default true)	Flag de estado. Permite habilitar o deshabilitar por completo el uso de esta ruta de conexión en el Job.
dn_usuario_creador	STRING	Sí	Auditoría: Usuario que registró los datos de esta conexión por primera vez.
dd_fecha_creacion	TIMESTAMP	No (Default Now)	Auditoría: Fecha y hora de creación de la conexión.
dn_usuario_modificador	STRING	Sí	Auditoría: Último usuario que modificó valores de red o credenciales.
dd_fecha_modificacion	TIMESTAMP	Sí	Auditoría: Última fecha de actualización del host, puerto o llaves.

---

2. Job Databricks

El Job automatizado que se ejecuta en el clúster de Databricks está alojado en el espacio de trabajo corporativo y depende de una arquitectura desacoplada en dos Notebooks escritos en Python: uno actúa como el orquestador principal del flujo y el otro centraliza las funciones utilitarias y la lógica pesada del pipeline.

---

2.1. Ubicación en el Workspace

• Workspace: dbw-alpura-lakehouse-dev-estatus2
• Carpeta del Proyecto: Catalogos Generales PostgreSQL
• Componentes:
  • 📑 Sync: Notebook principal invocado directamente por el Job de Databricks.
  • 📑 Replica Individual: Notebook principal invocado directamente la Replica de Catalogos de forma individual.
  • ⚙️ Utilities: Notebook secundario que almacena las constantes, conexiones JDBC, manejo de logs y lógica de réplica.

---

2.2. Notebook's

⚙️ 2.2.1 Orquestador: Sync

Este script contiene el punto de entrada oficial para la ejecución del Job. Su única responsabilidad es cargar las herramientas globales y disparar secuencialmente la réplica para cada uno de los tres entornos de Alpura.

---

⚙️ 2.2.2. Individual: Replica Individual

El notebook Utilities actúa como el motor lógico y núcleo operativo para la réplica de los catálogos de forma individual hacia Databricks. Este proceso permite la ejecución manual y controlada de vistas específicas según la capa de datos requerida.

---

⚙️ 2.2.3. Especificación de Funciones: Utilities

El notebook Utilities actúa como el motor lógico y núcleo operativo del pipeline de réplicas hacia PostgreSQL en GCP. Este archivo no se ejecuta de forma aislada, sino que provee todas las definiciones de constantes, validaciones de infraestructura, formateo de strings y el proceso de persistencia de datos (JDBC) al notebook orquestador Sync.

Constantes de Administración e Infraestructura

Al inicio del script, se definen los punteros absolutos hacia las tablas Delta de gobierno y auditoría dentro de Databricks:

admin_catalogos = "cat_master_aplicaciones.catalogos.catalogos"
admin_conexiones = "cat_master_aplicaciones.catalogos.conexiones"
admin_logs = "cat_master_aplicaciones.catalogos.logs"
admin_log_detail = "cat_master_aplicaciones.catalogos.log_details"

---

Catálogo de funciones dentro del archivo de Utilities

La siguiente tabla describe las funciones identificadas en el proceso de replicación replica_catalogo.

Función	Categoría	Parámetros principales	Retorno	Descripción	Se utiliza cuando
replica_catalogo	Orquestación	source, catalogo	None o resultado de ejecución	Función principal del proceso. Coordina la obtención de metadatos, validaciones, creación o actualización de tablas, sincronización incremental y auditoría.	Al iniciar la replicación de uno o varios catálogos.
table_exists	Validación	object	bool	Verifica mediante Spark SQL si una tabla o vista existe en Databricks.	Antes de consultar o procesar un objeto de Databricks.
get_name_target	Utilidad	name_target	str	Normaliza el nombre de la tabla destino. Elimina el sufijo _v y agrega el prefijo alp_cat_.	Al construir el nombre de la tabla destino en PostgreSQL.
get_data_tipe_postgres	Utilidad	dtype	str	Convierte tipos de datos de Spark o Databricks a tipos compatibles con PostgreSQL.	Durante la creación de tablas o al agregar columnas nuevas.
get_catalogs	Metadatos	source, catalogo	dict	Obtiene la configuración de los catálogos activos desde admin_catalogos, incluyendo origen, destino, capa y llave primaria.	Al inicio de la ejecución, antes de iterar los catálogos.
get_conn_target	Conexión	source, catalogo	dict	Obtiene las credenciales y parámetros de conexión a PostgreSQL asociados al catálogo.	Antes de realizar consultas, DDL o cargas hacia PostgreSQL.
get_exist_target	Validación de destino	source, catalogo	dict	Consulta information_schema.tables para determinar si la tabla destino existe y define la acción create o update.	Antes de decidir si la tabla debe crearse o actualizarse.
create_table	Creación de tabla	source, database, schema, catalogo	dict	Lee la tabla origen, convierte columnas a minúsculas, genera row_hash y crea la tabla en PostgreSQL mediante modo overwrite.	Cuando la tabla destino no existe.
validar_columnas	Evolución de esquema	source, database, schema, catalogo	dict	Compara las columnas de Databricks contra PostgreSQL y agrega columnas nuevas mediante ALTER TABLE ... ADD COLUMN.	Cuando la tabla destino ya existe.
actualizar_info	Sincronización incremental	source, database, schema, catalogo, primary_key	dict	Detecta registros nuevos o modificados mediante row_hash y ejecuta un INSERT ... ON CONFLICT DO UPDATE.	Después de validar estructura y llave primaria de una tabla existente.
create_pk_table	Gestión de PK	Datos de conexión, esquema, tabla, llave primaria	None o dict	Crea una restricción de llave primaria en PostgreSQL. Soporta llaves simples y compuestas.	Cuando existe una PK configurada pero no existe en la tabla destino.
drop_pk_table	Gestión de PK	Datos de conexión, esquema, tabla	None o dict	Busca y elimina la restricción de llave primaria existente en PostgreSQL.	Cuando la PK existente no coincide con la configurada.
val_pk_table	Validación de PK	Datos de conexión, esquema, tabla, llave primaria	dict	Valida que la PK de PostgreSQL coincida con la PK configurada; si difiere, elimina y recrea la restricción.	Antes de ejecutar la carga incremental.
log_inicio	Auditoría global	ejecucion_id, usuario, parámetros	None	Crea el registro inicial de la ejecución global.	Al inicio de replica_catalogo.
log_update_total_catalogos	Auditoría global	ejecucion_id, total de catálogos	None	Actualiza el número total de catálogos encontrados para la ejecución.	Después de obtener los catálogos mediante get_catalogs.
log_detail_inicio	Auditoría por catálogo	ejecucion_id, catálogo, datos de ejecución	None	Registra el inicio del procesamiento individual de una tabla.	Antes de procesar cada catálogo.
log_detail_existe_target	Auditoría por catálogo	ejecucion_id, catálogo, indicador de existencia	None	Registra si la tabla destino existe en PostgreSQL.	Después de ejecutar get_exist_target.
log_detail_existe_pk	Auditoría por catálogo	ejecucion_id, catálogo, indicador de PK	None	Registra si la tabla destino tiene una llave primaria existente.	Durante la validación de llave primaria.
log_detail_creando_pk	Auditoría por catálogo	ejecucion_id, catálogo, llave primaria	None	Registra el intento o resultado de creación de una llave primaria.	Cuando se ejecuta create_pk_table.
log_detail_columnas_nuevas	Auditoría por catálogo	ejecucion_id, catálogo, columnas	None	Registra las columnas nuevas detectadas y agregadas en PostgreSQL.	Cuando validar_columnas encuentra diferencias de esquema.
log_detail_totales	Auditoría por catálogo	ejecucion_id, catálogo, métricas	None	Registra métricas de registros iniciales, finales, cambios, inserciones y actualizaciones.	Después de crear o actualizar la tabla.
log_detail_fin	Auditoría por catálogo	ejecucion_id, catálogo, estado, fechas	None	Cierra el procesamiento de un catálogo y calcula su duración.	Al finalizar cada iteración de catálogo.
log_fin	Auditoría global	ejecucion_id, estado, error opcional	None	Finaliza la ejecución global con estado Ok o Error.	Al finalizar el proceso completo o cuando ocurre un error.

---

3. Dependencia y Carga de Módulos

Al inicio, el script realiza una llamada mágica de Databricks (%run) para importar en memoria todas las variables y funciones desarrolladas en el componente de utilerías:

# Carga de Utilerias
%run ./Utilities

# Sync de todos los catalogos (Ejecutado por JOB)
replica_catalogo("gold", None)
replica_catalogo("silver", None)
replica_catalogo("bronze", None)

# Replica de Catalogos Individuales (Petición Manual)
replicar_catalogo_individual("gold", "alp_cat_qp_matriz_items_precios_v")

---

4. Pasos de Ejecución Manual

Para ejecutar la réplica de un catálogo de manera individual, sigue estrictamente este orden:

---

4.1. Seleccionar el Clúster (Compute)

Asegúrate de que tu notebook esté adjunto al clúster de cómputo alpura-dev-etl-shared-lakehouse.

---

4.2. Cargar Utilerías:

Ejecuta la celda correspondiente a la Carga de Utilerías para inicializar todas las dependencias y funciones necesarias.

---

4.3. Configurar y Ejecutar:

En la función de Python encargada de la ejecución, modifica los parámetros Source y catalogo, y ejecuta el procedimiento.

---

5. Parámetros de Configuración de replica_catalogo

Para ejecutar el Procedimiento replica_catalogo se utilizan los siguiente parametros :

Parámetro	Obligatorio	Descripción
source	Y	Define la capa del Lakehouse de donde se extraerá la información.
catalogo	N	Nombre exacto del catálogo en databricks o vista a replicar.

Ejemplos de Ejecución

Sincronización Completa (Todos los catálogos en un ambiente)
Si deseas procesar y sincronizar de forma masiva todos los catálogos activos para un ambiente específico, pasa el parámetro del catálogo como None:

replicar_catalogo_individual("gold", None)

Sincronización Individual
Si deseas procesar y sincronizar de manera individual para un ambiente específico, pasa el nombre del catalogo:

replicar_catalogo_individual("gold", "alp_cat_ar_clientes_v")

---

5.1. Sources

Sources	Descripción
bronze	Información de Ambiente Pruebas de Oracle Cloud.
silver	Información de Ambiente UAT de Oracle Cloud.
gold	Información de Ambiente Productivo de Oracle Cloud.

---

5.2. Catalogos en Databricks

Catalogo	Descripción
alp_cat_ar_clientes_v	Catálogo de Clientes (Cuentas Financieras): Contiene el maestro de cuentas de clientes registrado en el módulo de Cuentas por Cobrar (AR) de Oracle, útil para la conciliación financiera y facturación.
alp_cat_cdm_clientes_v	Catálogo Maestro de Clientes (CDM): Centraliza la información del Customer Data Management, consolidando el registro único global (Party) de clientes antes de su segmentación transaccional.
alp_cat_fnd_flex_fields_v	Campos Flexibles (Flexfields): Almacena las configuraciones de campos personalizados definidos en Oracle Fusion para extender la información maestra de los registros.
alp_cat_fnd_lookups_v	Catálogo de Lookups (Valores de Referencia): Contiene las listas de valores, catálogos rápidos y traducciones técnicas utilizadas por Oracle para homogeneizar campos de estado, tipos y clasificaciones.
alp_cat_hcm_employees_v	Maestro de Empleados (HCM): Contiene la información general de la estructura de colaboradores y personal de Alpura extraída del módulo de Human Capital Management.
alp_cat_inv_almacenes_v	Catálogo de Almacenes de Inventario: Define el maestro de organizaciones de inventario, subinventarios y almacenes físicos u operativos dentro de la cadena de suministro de Alpura.
alp_cat_inv_items_v	Maestro de Artículos (Items): Contiene la definición maestra de productos, materiales, códigos de barra y atributos de inventario de todos los SKUs de Alpura.
alp_cat_inv_localizadores_v	Catálogo de Localizadores (Locators): Almacena las ubicaciones físicas detalladas (pasillos, estantes, racks) dentro de los almacenes de inventario de Alpura.
alp_cat_qp_discount_list_v	Listas de Descuentos (QP): Registra las reglas, porcentajes, montos y vigencias de los descuentos aplicables a nivel de artículo o cliente en el módulo de Advanced Pricing.
alp_cat_qp_matriz_items_precios_v	Matriz Pivotada de Precios por Canal: Vista optimizada que consolida y pivota el precio base de cada SKU (item_oracle) a lo largo de todos los canales de distribución de Alpura (Autoservicio, Detalle, Mayoristas, etc.).
alp_cat_qp_price_lists_v	Listas de Precios Maetras (QP): Contiene las cabeceras e ítems asociados a las distintas listas de precios bases configuradas en Oracle Advanced Pricing.
alp_cat_qp_pricing_assignments_strategy_v	Matriz de Asignación de Estrategias: Define bajo qué reglas dinámicas (canal, subcanal, unidad de negocio y segmento) se asigna una estrategia de precios específica.
alp_cat_qp_pricing_oracle_v	Consolidado de Estrategias de Precios: Vista maestra que unifica las estrategias de precios (Pricing Strategies) junto con sus listas de precios y descuentos asociadas en una sola estructura lógica.
alp_cat_qp_pricing_profiles_v	Perfiles de Precios de Clientes: Clasifica el valor estratégico de los clientes en función de su tamaño, ingresos potenciales, costo de servicio y calificación crediticia.
alp_cat_qp_pricing_segments_v	Matriz de Segmentación de Clientes: Mapea las reglas de negocio dinámicas para clasificar automáticamente a cada cliente dentro de un segmento de precios específico.

---

6. Flujo General del Proceso