Lleva tu base de datos a un Iceberg Data Lake (Parte 2)

La serie

Esta es la continuación de cómo replicar tu base de datos hacia un data lake basado en Iceberg. En la Parte 1 vimos la teoría, el cómo y el porqué. En esta segunda parte cubriré la implementación real, con código, comandos y aprendizajes al llevar todo esto a producción.

Prerrequisitos de la implementación

Los únicos prerrequisitos reales son contar con una instancia en la nube (o una solución on-prem) que te permita levantar un cluster de Kubernetes, tener una fuente compatible con Debezium Server y un almacenamiento BLOB como GCS o S3 para guardar los datos de tus tablas Iceberg. En este artículo doy las instrucciones para configurar todo lo anterior sobre GCP, excepto la fuente.

Ni siquiera hace falta mucho conocimiento de Kubernetes para implementar este ejemplo básico, porque te voy a dar todos los comandos que se ejecutan. Créeme, sí lo vas a necesitar más allá de este ejemplo básico, cuando quieras hacerlo a escala de producción, pero para este caso no necesitas ser un Kubestronaut.

Ten en cuenta que esto también se puede desplegar por otras vías distintas a Kubernetes. Si lo vas a correr en solitario en un cluster de Kubernetes, lo recomiendo por el ahorro en costos. Si solo vas a correr esta aplicación y no necesitas un cluster de Kubernetes completo, considera correrlo como un contenedor standalone en una plataforma de cómputo como GCE o EC2.

El plan de implementación

El plan para esta implementación sigue estos pasos:

1. Preparar tu fuente (Postgres en este caso)
2. Configurar los permisos en tu bucket/destino
3. Modificar el deployment de Kubernetes con tus valores
4. Desplegar el deployment de Kubernetes
5. Probarlo

Es un plan muy simple a propósito, y voy a ir marcando los puntos en los que se pueden hacer mejoras para uso en producción, más allá de este caso simple que planteo como prueba de concepto.

Como a menudo me dicen "The BigQuery Dude", voy a implementar esto en GCP, porque es mi terreno. Ten en cuenta que este ejemplo también es trivial de desplegar en AWS o Azure. Voy a desplegar Debezium Server en GKE, usando como fuente Postgres en Cloud SQL y tablas Iceberg en GCS, todo lo cual tiene sus equivalentes en AWS y Azure.

Uso la mayor cantidad de código agnóstico a la plataforma que puedo, pero algo será específico de GCP. Todo lo que no sean los comandos "gcloud" y los prefijos "gs://" de abajo se traslada directamente a cualquier variante de Kubernetes, almacenamiento y Postgres que exista en los otros proveedores.

Para los datos de ejemplo a lo largo de esta entrada uso la famosa base de datos Northwind, con la que muchos de mi generación aprendimos SQL. Puedes descargar el código desde GitHub aquí o ir directo al archivo SQL aquí. Está en el esquema public y estoy corriendo todo con un usuario llamado "my_user" para facilitar la búsqueda y el reemplazo, además de poder correrlo directo como prueba.

La implementación (replicación de la base de datos origen)

El primer paso es dejar lista tu base de datos. No voy a entrar en el detalle de la configuración, porque ya se ha escrito muchas, muchas veces. Voy a asumir que estás usando PostgreSQL en Cloud SQL de GCP para este caso de prueba. Si no estás usando PostgreSQL, revisa aquí cómo hacer estos pasos para tu base de datos preferida.

Primero, asegúrate de que tu flag wal_level esté en logical, sobre todo si son instancias autoalojadas o que no son Cloud SQL. En Cloud SQL, esto lo maneja la capa administrada y se llama flag "logical_decoding". Para activarlo, edita tu instancia y en flags verifica que cloudsql.logical_decoding esté en On; si no está agregado, agrégalo y asegúrate de que el valor quede en On.

Ahora toca configurar bien el usuario de replicación. Recomiendo ENCARECIDAMENTE crear un usuario nuevo para esto con un nombre descriptivo, para monitorearlo mejor. Suele llamarse debezium o cdc, pero puedes elegir el que prefieras.

Para crear un nuevo usuario, corre estos comandos (y revisa más abajo si te aparece un error):

CREATE USER my_user WITH REPLICATION LOGIN PASSWORD '<password>';

Si al ejecutar el comando anterior te aparecen errores relacionados con no poder agregar el rol REPLICATION a un usuario, corre este comando sobre tu usuario (asumiendo que tienes los privilegios; si no, quizás necesites que un admin lo haga por ti o usar el usuario postgres):

ALTER USER my_user WITH REPLICATION;

Ahora llega el momento de la verdad: elegir qué tabla quieres replicar. Aquí uso una sola tabla por simplicidad. Corre este comando, especificando las tablas y esquemas, y otorgando acceso al usuario elegido. Puedes especificar tantas como necesites, separadas por coma:

-- Cambia "my_publication" por el nombre que prefieras
CREATE PUBLICATION my_publication
FOR TABLE public.orders;

-- Corre esto para cada esquema seleccionado arriba
GRANT SELECT ON ALL TABLES IN SCHEMA public TO my_user;
GRANT USAGE ON SCHEMA public TO my_user;
ALTER DEFAULT PRIVILEGES IN SCHEMA public
    GRANT SELECT ON TABLES TO my_user;

Para una aplicación más práctica que la que cubro en este artículo, esta es la sintaxis SQL para hacerlo con varias tablas o con todas las tablas de un esquema:

-- Para varias tablas
CREATE PUBLICATION my_publication
FOR TABLE public.orders, public.order_details;

-- Para todas las tablas de un esquema
CREATE PUBLICATION my_publication
FOR TABLES IN SCHEMA public;

-- Corre esto para cada esquema seleccionado arriba
GRANT SELECT ON ALL TABLES IN SCHEMA public TO my_user;
GRANT USAGE ON SCHEMA public TO my_user;
ALTER DEFAULT PRIVILEGES IN SCHEMA public GRANT SELECT ON TABLES TO my_user;

Por último, hay que crear un replication slot. Un replication slot no es más que un proceso que corre de forma constante, publicando los datos modificados a cualquier receptor. Nota que lo llamo my_replication_slot para que sea más fácil ejecutarlo y hacer búsqueda/reemplazo más adelante. Para crearlo, corre el siguiente comando:

SELECT PG_CREATE_LOGICAL_REPLICATION_SLOT('my_replication_slot', 'pgoutput');

En este punto ya terminaste con la configuración de la replicación en la base de datos origen (asumiendo que estás siguiendo el ejemplo con Postgres).

Nota importante: Si no estás usando este replication slot de forma activa, va a correr continuamente y a almacenar los datos modificados en disco. Así que si no lo estás usando, elimínalo, porque sí va a empezar a consumir tu espacio en disco con el tiempo, lo que se traduce en una instancia de base de datos más costosa.

Notas de seguridad sobre la cuenta de servicio

En este ejemplo estoy corriendo el cluster de GKE, y por lo tanto el servicio de Debezium Server, con la cuenta de servicio por defecto de Compute Engine. Es solo para fines demostrativos, para no tener que entrar en el Principio de Mínimo Privilegio. NO hagas esto en producción con la cuenta de servicio por defecto. Crea una cuenta de servicio nueva que tenga solo los permisos necesarios; profundizar en esto queda fuera del alcance de este artículo. Te recomiendo iniciar el camino por esta madriguera con este artículo en la documentación de Google.

Además, verás que abajo señalo los lugares donde hago cosas por simplicidad y no por seguridad, ya que es una demostración. Toma nota, porque te estoy ahorrando dolores de cabeza de seguridad más adelante, y usa un método más seguro. ¡POR FAVOR!

La implementación (arma tus variables de entorno)

Agrego este paso para hacerte la vida más fácil, porque no lo hice la primera vez cuando escribí esto, y al revalidar todo antes de publicar, lo incorporé para hacerla MUCHO más sencilla. Este paso hace que los comandos de abajo sean más fáciles, porque ya no tienes que editar tus valores en cada uno. No modifiques nada después de la línea punteada, porque esos valores se generan automáticamente.

Nota que aquí hay credenciales, así que si vas a hacer esto en producción, por favor usa un método más seguro, como un secret manager.

El último conjunto de comandos imprimirá cada valor para validarlo, salvo el de la contraseña. Si ves errores, revisa tus valores y asegúrate de haberlos configurado correctamente.

Aquí está el comando con comentarios sobre cada uno para decirte qué hace:

# El nombre del proyecto
PROJECT_NAME=<project_name>
# El nombre del bucket donde vivirán las tablas Iceberg
BUCKET_NAME=<bucket-name>
# Ruta base en el bucket anterior donde vivirán las tablas Iceberg,
# normalmente es un nombre de tabla o esquema.
# Si no estás seguro, déjalo vacío y se escribirá en la raíz
# del bucket.
BUCKET_PATH=<bucket_path>
# Nombre de tu cluster; yo uso cdc_cluster en los comandos de arriba
CLUSTER_NAME=<cluster_name>
# Región del cluster; yo uso us-central1 en los comandos de arriba.
# Debería ser la misma región que tu instancia de Cloud SQL
CLUSTER_REGION=<cluster_region>
# Nombre de la instancia de Cloud SQL
SQL_INSTANCE_NAME=<cloud_sql_instance_name>
# Nombre de usuario de la base de datos que creaste arriba
DATABASE_USER=<database_user>
# Contraseña de la base de datos para el usuario que creaste arriba
DATABASE_PASSWORD=<database_password>
# Nombre de la base de datos donde se aplicaron los comandos anteriores
DATABASE_NAME=<database_name>

#----------------------------------------------------------------------
# NO modifiques nada debajo de esta línea, porque estos valores obtienen
# datos adicionales y llenan variables de entorno generadas para uso posterior.
gcloud config set project $PROJECT_NAME
PROJECT_NAME=$(gcloud config get-value project)
PROJECT_ID=$(gcloud config get-value project 2>/dev/null)
SQL_NAME=gcloud sql instances describe $SQL_INSTANCE_NAME --format='value(connectionName)'

# Lo siguiente imprimirá todo lo anterior para verificar que se ve correcto
echo "Project Name: $PROJECT_NAME"
echo "Project ID (Generated from gcloud): $PROJECT_ID"
echo "Bucket Name: $BUCKET_NAME"
echo "Bucket Path: $BUCKET_PATH"
echo "Full Bucket Path: gs://$BUCKET_NAME/$BUCKET_PATH"
echo "Cluster Name: $CLUSTER_NAME"
echo "Cluster Region: $CLUSTER_REGION"
echo "SQL Instance Name (Given): $SQL_INSTANCE_NAME"
echo "SQL Instance Name (Received from gcloud): $SQL_NAME"
echo "Database Username: $DATABASE_USER"
echo "Database Password (hidden): ******"
echo "Database Name: $DATABASE_NAME"

La implementación (permisos del bucket)

Para poder leer y escribir en un bucket de GCS, primero hay que crear el bucket. Luego hay un conjunto de permisos que se deben otorgar a la cuenta de servicio que corre el cluster de GKE que implementaremos en la siguiente sección.

He probado y tratado de reducir los requisitos de IAM para que sean más simples que lo que recomienda Google, pero en mis pruebas los siguientes dos roles predefinidos son las mejores opciones para otorgar a esta cuenta de servicio:

• Storage Object Admin (es decir, roles/storage.objectAdmin)
• Storage Legacy Bucket Reader (es decir, roles/storage.legacyBucketReader)

Google menciona esto para su implementación de BigLake, y recomiendo leer esa página si vas a usar BigLake. El detalle que ahí no aparece es que vas a necesitar los permisos del rol Storage Object Admin para escribir tablas Iceberg. Simplemente aplica ese rol y no te metas en la madriguera de errores y de probar combinaciones de permisos; hazme caso.

Corre el siguiente comando para crear el bucket y aplicar los bindings de la política de IAM a la cuenta de servicio, usando la de Compute Engine por defecto, así que cámbialo según lo necesites:

gcloud storage buckets create gs://$BUCKET_NAME
gcloud storage buckets add-iam-policy-binding gs://$BUCKET_NAME \
    --member="serviceAccount:$PROJECT_NAME[email protected]" \
    --role="roles/storage.objectAdmin"
gcloud storage buckets add-iam-policy-binding gs://$BUCKET_NAME \
    --member="serviceAccount:$PROJECT_NAME[email protected]" \
    --role="roles/storage.legacyBucketReader"

Los comandos son fáciles de entender. El primero solo apunta tu gcloud al proyecto correcto. El segundo crea el bucket. Los dos últimos otorgan los permisos necesarios a la cuenta de servicio por defecto de compute en el proyecto.

La implementación (creación de un cluster de Kubernetes)

Esta implementación consiste en crear un cluster de Kubernetes con Google Kubernetes Engine (GKE) Autopilot y cargar Debezium Server sobre él mediante un Helm chart. Si ya te manejas con GKE o tienes otro cluster corriendo, probablemente sepas lo que haces, así que puedes saltar directo a los comandos para cargar esto más abajo.

Primero, hay que agregar un rol a nuestra cuenta de servicio por defecto de Compute Engine, que es necesario. Aquí está el comando:

gcloud projects add-iam-policy-binding $PROJECT_ID \
    --member="serviceAccount:$PROJECT_NAME[email protected]" \
    --role="roles/container.defaultNodeServiceAccount"

A continuación, creamos el cluster de GKE Autopilot. Solo corre el comando de abajo.

gcloud container clusters create-auto $CLUSTER_NAME \
    --location=$CLUSTER_REGION \
    --project=$PROJECT_ID

Esto tardará un poco en levantarse, pero después de unos minutos deberías recibir la confirmación de que se completó, y podrás ver tu cluster en la sección de GKE de la Consola de GCP.

Nota sobre networking para aplicaciones del mundo real

Al correr esto en el mundo real, hay buenas probabilidades de que estés ejecutando Cloud SQL en otra región o lidiando con comunicación entre proyectos. Mi mayor recomendación es que hagas lo posible por usar la IP privada y el Cloud SQL proxy para sortear algunas de las "particularidades" del networking en GCP.

La implementación (despliegue en el cluster de Kubernetes)

Ahora viene la parte divertida: desplegar el Debezium Operator en GKE. Solo corre este comando y dale unos minutos para que descargue y se cargue:

# Autentica gcloud con el cluster de GKE; esto permite correr kubectl más adelante
gcloud container clusters get-credentials $CLUSTER_NAME --region $CLUSTER_REGION

# Agrega el repositorio Helm y actualiza la lista local de fuentes Helm
helm repo add debezium https://charts.debezium.io
helm repo update

# Instala el operator en un namespace dedicado (por ejemplo, debezium-system)
# Creará este namespace si no existe
helm install debezium-operator debezium/debezium-operator \
  --namespace debezium-system \
  --create-namespace

Nota: dejé los comentarios en el comando para explicar qué hace cada uno, así te ayudarán a entender lo que ocurre y a diagnosticar posibles problemas.

Para validar que se desplegó y/o diagnosticar por qué falló el despliegue en el peor de los casos, corre el comando de abajo, que lista todos los pods en el namespace creado y también muestra los logs asociados:

# Obtiene los pods en el namespace debezium-system
kubectl get debeziumserver -n debezium-system
# Muestra los logs de los pods
kubectl logs -l app.kubernetes.io/name=debezium-server -n debezium-system

¡Ya estás listo para hacer el despliegue real!

El comando de abajo es un poco largo, pero está haciendo una operación relativamente simple: desplegar Debezium Server y su configuración con un Cloud SQL proxy adjunto. Para quienes tienen más soltura con Kubernetes, está desplegando el Cloud SQL Proxy como sidecar en un pod que corre Debezium Server.

Cuando estés listo, corre este comando:

kubectl apply -f - <<EOF
apiVersion: debezium.io/v1alpha1
kind: DebeziumServer
metadata:
  name: debezium-sql-proxy
  namespace: debezium-system
spec:
  image: debezium/server:latest
  # --- Configuración del Sidecar ---
  runtime:
    containers:
      - name: cloud-sql-proxy
        image: gcr.io/cloud-sql-connectors/cloud-sql-proxy:latest
        args:
          - "--port=5432"
          - "$PROJECT_NAME:$CLUSTER_REGION:$SQL_NAME"
        securityContext:
          runAsNonRoot: true
  # --- Configuración del Conector ---
  source:
    connector.class: io.debezium.connector.postgresql.PostgresConnector
    config:
      # Conectar a localhost porque el proxy está en el mismo Pod
      database.hostname: "127.0.0.1"
      database.port: "5432"
      database.user: "$DATABASE_USER"
      database.password: "$DATABASE_PASSWORD"
      database.dbname: "$DATABASE_NAME"
      topic.prefix: "cdc"
      # Requerido para Cloud SQL Postgres
      plugin.name: "pgoutput"
  # --- Configuración del Sink de Iceberg ---
  sink:
    type: iceberg
    config:
      # Tipo de catálogo Iceberg (Hadoop es el mejor para setups solo con GCS)
      debezium.sink.iceberg.catalog.type: hadoop
      debezium.sink.iceberg.catalog.warehouse: gs://$BUCKET_NAME/$BUCKET_PATH

      # Configuración del FileSystem de GCS
      debezium.sink.iceberg.fs.defaultFS: gs://$BUCKET_NAME
      debezium.sink.iceberg.io-impl: org.apache.iceberg.gcp.gcs.GCSFileIO

      # Comportamiento de la tabla: upsert (fusiona cambios) o append
      debezium.sink.iceberg.upsert: "true"
      debezium.sink.iceberg.upsert.keep-deletes: "false"

      # Intervalo de commit (cada cuánto se crea un nuevo snapshot de Iceberg)
      debezium.sink.iceberg.table.commit.interval-ms: "60000"
EOF

Probando el resultado (¡por fin!)

¡Llegó el momento de probar! Ahora entra a tu Postgres e inserta filas en la tabla incluida en el comando que corrimos antes. Eso es todo lo que hay que hacer del lado de Postgres para disparar la replicación en adelante. Debería empezar a escribir datos directamente en una tabla Iceberg en GCS.

Para probarlo, vamos a usar BigQuery para consultar la tabla Iceberg. En BigQuery vamos a crear una BigLake External Table. En términos no técnicos de BigQuery, esto significa que le estás diciendo a BigQuery: tengo tablas en un formato específico que viven en un bucket de GCS y me gustaría consultarlas desde BigQuery vía una capa de abstracción llamada BigLake para las lecturas en GCS.

Para hacerlo, abre tu Consola de GCP y navega a la sección de BigQuery, que debería abrir una ventana de editor de texto en lo que llaman BigQuery Studio.

Pega este comando ahí y modifícalo con la región donde vive tu bucket, el nombre del bucket y la ruta del bucket que definiste arriba:

CREATE SCHEMA IF NOT EXISTS `testing_biglake_dataset`
OPTIONS(
  location = '<bucket_region>'
);

CREATE OR REPLACE EXTERNAL TABLE `testing_biglake_dataset.my_testing_iceberg_table`
WITH CONNECTION `testing_biglake_dataset.my_connection`
OPTIONS (
  format = 'ICEBERG',
  uris = ['gs://<bucket_name>/<bucket_path>/metadata.json']
);

Adelante, córrelo; creará un dataset específico nuevo para esto. Así podemos eliminarlo todo después sin complicaciones.

Cuando termine esa consulta, haz clic en la pestaña de nueva consulta en la parte superior del panel del editor de BigQuery Studio. Luego pega este SQL y córrelo:

SELECT * FROM testing_biglake_dataset.my_testing_iceberg_table TABLESAMPLE SYSTEM (15 PERCENT)

Esto debería devolver algunas filas de los datos que insertaste. Ten en cuenta que hace un muestreo del 15% de la tabla, así que no devolverá todos los datos. Como estamos hablando de BigQuery, y de su increíble habilidad para generar facturas enormes, tomo esta precaución para que las consultas sean muy pequeñas.

¡En este punto ya has implementado con éxito la replicación de datos de Postgres a Iceberg! Lo siguiente es hacer lo mismo para varias tablas (o esquemas) y tener así todos tus datos en un almacenamiento "agnóstico a la base de datos".

En conclusión

Esto es apenas la escritura básica de una sola tabla hacia una sola tabla Iceberg, pero el potencial es enorme. Imagina poder escribir bases de datos de ventas enteras, llenas de transacciones, en tiempo real hacia un data warehouse, que luego se puede consultar para reflejar datos de ventas actualizados en dashboards y reportes.

Lo mejor es que el data warehouse que hace la consulta no importa; si prefieres BigQuery, ClickHouse, Snowflake, Databricks, etc., o una mezcla, no hay problema. Mantenerse agnóstico a la plataforma es una de las mejores prácticas que puedes adoptar hoy para seguir siendo competitivo, tanto técnicamente como para llevar la delantera en costos.

Este es solo uno de los muchos temas que cubrimos y en los que asesoramos en DoiT International, así que si te interesa saber más, contáctanos para ver cómo podemos ayudarte con tus costos en la nube o con tus operaciones en la nube.

Limpieza (si/cuando quieras)

Por último, si después de probar esto quieres eliminarlo para reducir costos y planear cómo correrlo en producción, aquí está el comando para "quemarlo todo" en el cluster de GKE:

gcloud container clusters delete $CLUSTER_NAME \
    --location=$CLUSTER_REGION \
    --project=$PROJECT_ID

Para deshacerte del dataset de BigQuery que creamos, corre esto en BigQuery Studio:

DROP SCHEMA IF EXISTS `testing_biglake_dataset`;

NO olvides este paso si vas a eliminar todo, ya que esto quita el replication slot de tu instancia de Postgres. Si no lo haces, los costos de almacenamiento en Postgres van a crecer con el tiempo. Ten en cuenta que este comando se corre dentro de Postgres:

SELECT pg_drop_replication_slot('my_replication_slot');

Cómo lo hacemos en DoiT

Aquí en DoiT International resolvemos problemas como este todo el tiempo, y siempre me preguntan cómo ayudar a ahorrar dinero al implementar proyectos de datos.

Ayudar a implementar proyectos de la forma más efectiva y más eficiente en costos es parte de nuestra misión con los clientes. Manejar todo, desde casos como estos hasta brindar las mejores soluciones de FinOps para nuestros clientes, es lo que hacemos, y quizás sea un poco parcial al decirlo, pero lo hacemos MUY bien.

Además, ofrecemos herramientas de primer nivel para tus necesidades de optimización de costos, como PerfectScale para Kubernetes y Select para BigQuery, Databricks y Snowflake,