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:
cat_master_aplicaciones.catalogos.catalogos: Define qué tablas/vistas se replican y a qué ambientes aplican.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. |