Portare il database su un data lake Iceberg (Parte 2)

La serie

Questo articolo prosegue il tema della replica del database verso un data lake basato su Iceberg. Nella Parte 1 abbiamo affrontato la teoria, il come e il perché di questa scelta. In questa seconda parte entrerò nel merito dell'implementazione vera e propria, con codice, comandi e lezioni apprese portandola in produzione.

I prerequisiti per l'implementazione

Gli unici veri prerequisiti sono disporre di un'istanza cloud (o di una soluzione on-prem) su cui avviare un cluster Kubernetes, avere una sorgente compatibile con Debezium Server e uno store BLOB come GCS o S3 in cui memorizzare i dati delle tabelle Iceberg. In questo articolo fornirò le istruzioni per configurare tutto — tranne la sorgente — su GCP.

Non serviranno grandi conoscenze di Kubernetes per riprodurre questo esempio di base, perché fornirò tutti i comandi da eseguire. Credetemi, andando oltre l'esempio base e portando il tutto in produzione a pieno regime serviranno eccome, ma per questo scenario non c'è bisogno di essere dei Kubestronauti.

Tenete presente che il deploy si può fare anche in modi diversi da Kubernetes: se dovete eseguire questa applicazione da sola su un cluster Kubernetes, per contenere i costi consiglio l'approccio qui descritto. Se invece dovete eseguire soltanto questa applicazione, senza avere realmente bisogno di un cluster Kubernetes completo, valutate l'esecuzione come container standalone su una piattaforma di compute come GCE o EC2.

Il piano di implementazione

Il piano per questa implementazione segue questi passaggi:

1. Preparare la sorgente (in questo caso Postgres)
2. Configurare i permessi sul bucket/destinazione
3. Modificare il deployment Kubernetes con le vostre impostazioni
4. Effettuare il deploy su Kubernetes
5. Testare il risultato

È un piano volutamente molto semplice; nei punti opportuni segnalerò dove si può migliorare per un uso in produzione che vada oltre questo caso d'uso semplice, pensato come proof-of-concept.

Dato che mi chiamano spesso "The BigQuery Dude", implementerò tutto su GCP, il mio terreno di casa. Va detto che questo esempio è banale da riprodurre anche su AWS o Azure. Utilizzerò Debezium Server su GKE, Postgres su Cloud SQL come sorgente e tabelle Iceberg su GCS: tutti servizi con equivalenti su AWS e Azure.

Ho cercato di usare codice il più possibile agnostico rispetto alla piattaforma, ma qualcosa sarà specifico di GCP. Tutto ciò che non riguarda i comandi "gcloud" e i prefissi "gs://" qui sotto è trasferibile direttamente a qualsiasi variante di Kubernetes, storage e Postgres disponibile sugli altri provider.

Come dati di esempio userò il famoso database Northwind, quello su cui molti della mia generazione hanno imparato SQL. Il codice si trova su GitHub qui, oppure andate direttamente al file SQL qui. Il tutto risiede nello schema public ed eseguirò ogni comando con un utente chiamato "my_user", in modo da rendere agevoli le operazioni di search & replace e poter lanciare i comandi direttamente come test.

L'implementazione (replica del database sorgente)

Il primo passo è preparare il database. Non entrerò nei dettagli della configurazione, perché se ne è già scritto tantissime volte. Presumo che stiate utilizzando PostgreSQL su Cloud SQL di GCP per questo test. Se non usate PostgreSQL, date un'occhiata qui per gli stessi passaggi sul database di vostra scelta.

Per prima cosa, assicuratevi che il flag wal_level sia impostato su logical, soprattutto per istanze self-hosted o non Cloud SQL. Su Cloud SQL se ne occupa il livello gestito tramite il flag "logical_decoding". Per aggiungerlo, modificate l'istanza e, nella sezione dei flag, verificate che cloudsql.logical_decoding sia impostato su On; se non è presente, aggiungetelo con valore On.

Passiamo poi alla configurazione dell'utente di replica. Consiglio VIVAMENTE di creare un nuovo utente dedicato, con un nome descrittivo, così da poterlo monitorare meglio. Di solito lo vedo chiamato debezium o cdc, ma potete scegliere il nome che preferite.

Per creare un nuovo utente, eseguite questi comandi (e vedete più sotto in caso di errori):

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

Se eseguendo il comando precedente ricevete errori relativi all'impossibilità di aggiungere il ruolo REPLICATION a un utente, eseguite questo comando sull'utente (a patto di averne i privilegi; in caso contrario dovrete chiedere a un amministratore o usare l'utente postgres):

ALTER USER my_user WITH REPLICATION;

È il momento della verità: scegliere quale tabella replicare. Per semplicità ne uso una sola. Eseguite il comando seguente, indicando tabelle e schemi e assegnando all'utente selezionato i permessi necessari. Potete indicarne quanti ne servono, separati da una virgola:

-- Sostituite "my_publication" con il nome che preferite
CREATE PUBLICATION my_publication
FOR TABLE public.orders;

-- Da eseguire per ciascuno schema selezionato sopra
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;

Per un'applicazione più concreta di quella trattata in questo articolo, ecco la sintassi SQL per gestire più tabelle o tutte le tabelle di uno schema:

-- Per più tabelle
CREATE PUBLICATION my_publication
FOR TABLE public.orders, public.order_details;

-- Per tutte le tabelle di uno schema
CREATE PUBLICATION my_publication
FOR TABLES IN SCHEMA public;

-- Da eseguire per ciascuno schema selezionato sopra
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;

Infine dobbiamo creare uno slot di replica. Uno slot di replica è semplicemente un processo sempre attivo che pubblica i dati modificati verso i vari destinatari. Nell'esempio lo chiamo my_replication_slot, per semplificare esecuzione e search & replace successivi. Per crearlo, eseguite il comando seguente:

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

A questo punto la configurazione della replica sul database sorgente è completa (sempre che stiate seguendo con Postgres).

Nota importante: se non state utilizzando attivamente questo slot di replica, continuerà comunque a funzionare senza sosta e a memorizzare su disco i dati modificati. Quindi, se non lo usate, cancellatelo: inizierà a consumare spazio su disco nel tempo, facendo lievitare i costi dell'istanza del database.

Note di sicurezza sui service account

In questo esempio eseguo il cluster GKE — e quindi il servizio Debezium Server — con il service account predefinito di Compute Engine. È solo a scopo dimostrativo, per non entrare nel merito del Principio del Minimo Privilegio. NON fatelo in produzione con il service account predefinito. Create un nuovo service account con soltanto i permessi effettivamente necessari; approfondire questo aspetto esula dallo scopo dell'articolo. Vi consiglio di iniziare a esplorare l'argomento partendo dalla relativa pagina della documentazione Google.

Più avanti segnalo anche i punti in cui procedo per semplicità e non per sicurezza, trattandosi di una dimostrazione. Prendetene nota: vi sto risparmiando grattacapi di sicurezza in futuro, quindi adottate un metodo più sicuro. PER FAVORE!

L'implementazione (definizione delle variabili d'ambiente)

Aggiungo questo passaggio per rendervi la vita più semplice: non l'avevo fatto la prima volta e, rivalidando tutto prima della pubblicazione, ho inserito questo step che semplifica ENORMEMENTE il lavoro. Serve a rendere i comandi qui sotto più agevoli, evitandovi di modificare i valori a ogni singolo passaggio. Non modificate nulla oltre la riga tratteggiata: quei valori vengono generati automaticamente.

Attenzione: qui sono presenti delle credenziali; se lo fate in produzione, usate un metodo più sicuro, ad esempio un secret manager.

L'ultimo blocco di comandi stamperà ogni valore per verifica, tranne la password. Se notate errori, controllate i valori e assicuratevi di averli impostati correttamente.

Ecco il comando, con un commento sopra ogni riga che ne spiega il significato:

# Nome del progetto
PROJECT_NAME=<project_name>
# Nome del bucket in cui risiederanno le tabelle Iceberg
BUCKET_NAME=<bucket-name>
# Percorso base nel bucket sopra indicato in cui risiederanno le tabelle Iceberg,
# in genere è il nome di una tabella o di uno schema.
# In caso di dubbio, lasciatelo vuoto: scriverà nella root del bucket.
BUCKET_PATH=<bucket_path>
# Nome del cluster; nei comandi sopra sto usando cdc_cluster
CLUSTER_NAME=<cluster_name>
# Regione del cluster; nei comandi sopra sto usando us-central1.
# Deve essere la stessa regione dell'istanza Cloud SQL
CLUSTER_REGION=<cluster_region>
# Nome dell'istanza Cloud SQL
SQL_INSTANCE_NAME=<cloud_sql_instance_name>
# Nome utente del database creato in precedenza
DATABASE_USER=<database_user>
# Password dell'utente del database creato in precedenza
DATABASE_PASSWORD=<database_password>
# Nome del database in cui sono stati applicati i comandi precedenti
DATABASE_NAME=<database_name>

#----------------------------------------------------------------------
# NON modificate nulla sotto questa riga: questi valori recuperano
# ulteriori informazioni e popolano variabili d'ambiente generate per l'uso successivo.
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)'

# Il blocco seguente stampa tutti i valori per verificare che siano corretti
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"

L'implementazione (permessi del bucket)

Per leggere e scrivere su un bucket GCS occorre innanzitutto creare il bucket. Servono poi alcuni permessi da concedere al service account con cui gira il cluster GKE che imposteremo nella prossima sezione.

Ho provato a semplificare i requisiti IAM rispetto a quanto suggerito da Google, ma dai miei test le due opzioni migliori come ruoli predefiniti da assegnare a questo service account sono:

• Storage Object Admin (ovvero roles/storage.objectAdmin)
• Storage Legacy Bucket Reader (ovvero roles/storage.legacyBucketReader)

Google ne parla nella pagina dedicata a BigLake, e ne consiglio la lettura se avete intenzione di usare BigLake. Il dettaglio che tralasciano è che servono i permessi del ruolo Storage Object Admin per scrivere le tabelle Iceberg. Assegnate direttamente questo ruolo e non perdetevi nel labirinto di errori tentando combinazioni di permessi: fidatevi.

Eseguite il comando seguente per creare il bucket e applicare le associazioni della policy IAM del service account (nell'esempio uso quello predefinito di Compute Engine; modificate secondo necessità):

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"

I comandi sono facili da capire. Il primo imposta gcloud sul progetto corretto. Il secondo crea il bucket. Gli ultimi due concedono i permessi necessari al service account Compute predefinito del progetto.

L'implementazione (creazione di un cluster Kubernetes)

Questa implementazione prevede la creazione di un cluster Kubernetes tramite Google Kubernetes Engine (GKE) Autopilot e il deploy di Debezium Server tramite Helm chart. Se avete già familiarità con GKE o disponete di un cluster già in esecuzione, probabilmente sapete già come muovervi e potete saltare direttamente ai comandi qui sotto.

Per prima cosa dobbiamo aggiungere un ruolo al service account predefinito di Compute Engine. Ecco il comando:

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

Passiamo ora alla creazione del cluster GKE Autopilot vero e proprio. Eseguite il comando qui sotto.

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

Ci vorrà un po' per l'avvio, ma dopo qualche minuto dovreste ricevere la conferma del completamento e vedere il cluster nella sezione GKE della GCP Console.

Nota sul networking per applicazioni reali

In scenari reali è molto probabile che Cloud SQL giri in un'altra regione o che si debba gestire la comunicazione tra progetti diversi. Il mio consiglio principale è sfruttare il più possibile l'IP privato e il Cloud SQL Proxy, così da aggirare alcune "peculiarità" del networking su GCP.

L'implementazione (deploy sul cluster Kubernetes)

Arriviamo alla parte divertente: il deploy dell'operatore Debezium su GKE. Eseguite questo comando e attendete qualche minuto per il download e il caricamento:

# Autentica gcloud con il cluster GKE, così potrete usare kubectl in seguito
gcloud container clusters get-credentials $CLUSTER_NAME --region $CLUSTER_REGION

# Aggiunge il repository Helm e aggiorna l'elenco locale dei sorgenti Helm
helm repo add debezium https://charts.debezium.io
helm repo update

# Installa l'operatore in un namespace dedicato (es. debezium-system)
# Verrà creato se non esiste già
helm install debezium-operator debezium/debezium-operator \
  --namespace debezium-system \
  --create-namespace

Nota: ho lasciato i commenti nel comando per spiegare cosa fa ciascuna riga, in modo da facilitare la comprensione e la diagnosi di eventuali problemi.

Per verificare che il deploy sia riuscito e, nel peggiore dei casi, diagnosticare l'errore, eseguite il comando seguente: elenca tutti i pod del namespace creato e ne mostra i log:

# Elenca i pod nel namespace debezium-system
kubectl get debeziumserver -n debezium-system
# Mostra i log dei pod
kubectl logs -l app.kubernetes.io/name=debezium-server -n debezium-system

Ora siete pronti per il deploy vero e proprio!

Il comando qui sotto è un po' lungo, ma svolge un'operazione relativamente semplice: fa il deploy di Debezium Server e della sua configurazione, con un Cloud SQL Proxy collegato. Per chi ha maggior dimestichezza con Kubernetes: si tratta di distribuire il Cloud SQL Proxy come sidecar in un pod su cui gira Debezium Server.

Quando siete pronti, eseguite semplicemente questo comando:

kubectl apply -f - <<EOF
apiVersion: debezium.io/v1alpha1
kind: DebeziumServer
metadata:
  name: debezium-sql-proxy
  namespace: debezium-system
spec:
  image: debezium/server:latest
  # --- Sidecar Configuration ---
  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
  # --- Connector Configuration ---
  source:
    connector.class: io.debezium.connector.postgresql.PostgresConnector
    config:
      # Connessione a localhost perché il proxy è nello stesso 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"
      # Necessario per Cloud SQL Postgres
      plugin.name: "pgoutput"
  # --- Iceberg Sink Configuration ---
  sink:
    type: iceberg
    config:
      # Tipo di catalogo Iceberg (Hadoop è la scelta migliore per setup solo GCS)
      debezium.sink.iceberg.catalog.type: hadoop
      debezium.sink.iceberg.catalog.warehouse: gs://$BUCKET_NAME/$BUCKET_PATH

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

      # Comportamento della tabella: upsert (unisce le modifiche) o append
      debezium.sink.iceberg.upsert: "true"
      debezium.sink.iceberg.upsert.keep-deletes: "false"

      # Intervallo di commit (ogni quanto creare un nuovo snapshot Iceberg)
      debezium.sink.iceberg.table.commit.interval-ms: "60000"
EOF

Testare l'output (finalmente!)

Ecco il momento del test! Accedete a Postgres e inserite qualche riga nella tabella coinvolta dal comando eseguito in precedenza. È tutto ciò che dovete fare lato Postgres per innescare la replica da questo momento in poi. I dati dovrebbero essere scritti direttamente in una tabella Iceberg su GCS.

Per verificarlo useremo BigQuery per interrogare la tabella Iceberg. In BigQuery creeremo una BigLake External Table. In parole povere, significa dire a BigQuery: "ho delle tabelle in un formato specifico che risiedono su un bucket GCS, e vorrei interrogarle tramite BigQuery grazie a un livello di astrazione, chiamato BigLake, che gestisce la lettura da GCS".

Aprite quindi la GCP Console e portatevi nella sezione BigQuery: si aprirà una finestra di editor testuale, in quello che chiamano BigQuery Studio.

Incollate lì il comando seguente, modificandolo con la regione del bucket, il nome del bucket e il percorso del bucket impostati in precedenza:

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']
);

Eseguitelo: verrà creato un dataset dedicato. In questo modo potremo rimuovere tutto in un secondo momento senza complicazioni.

Dopo l'esecuzione della query, cliccate sulla scheda per creare una nuova query in cima al pannello dell'editor di BigQuery Studio. Incollate quindi questo SQL ed eseguitelo:

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

Dovrebbe restituire alcune delle righe che avete inserito. Attenzione: viene eseguito un table sample al 15%, quindi non verranno restituiti tutti i dati. Trattandosi di BigQuery, con la sua notevole capacità di far lievitare i costi, adotto questa precauzione per far girare query molto contenute.

A questo punto avete implementato con successo la replica dei dati da Postgres a Iceberg! Il passo successivo è replicare l'operazione per più tabelle (o schemi), così da avere tutti i dati in un datastore "agnostico rispetto al database".

In conclusione

Questa è solo una configurazione di base che scrive una singola tabella su una singola tabella Iceberg, ma le potenzialità sono enormi. Immaginate di poter scrivere in tempo reale interi database di vendita, ricchi di transazioni, in un data warehouse che possa poi essere interrogato per riflettere dati di vendita aggiornati su dashboard e report.

Il bello è che il data warehouse che effettua le query non fa più differenza: BigQuery, ClickHouse, Snowflake, Databricks e così via — o una loro combinazione — vanno tutti bene. Restare agnostici rispetto alla piattaforma è oggi una delle best practice più utili per rimanere competitivi, sia sul piano tecnico sia sul controllo dei costi.

Questo è solo uno dei tanti temi su cui interveniamo e diamo consulenza in DoiT International: se volete approfondire, contattateci per scoprire come possiamo aiutarvi sui costi o sulle operations cloud.

Pulizia (se e quando vorrete)

Infine, se dopo i test volete rimuovere tutto per ridurre i costi e pianificare il passaggio in produzione, ecco il comando per "radere al suolo" il cluster GKE:

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

Per eliminare il dataset BigQuery creato, eseguite questo in BigQuery Studio:

DROP SCHEMA IF EXISTS `testing_biglake_dataset`;

NON dimenticate questo passaggio se state facendo pulizia: rimuove lo slot di replica dalla vostra istanza Postgres. Se non lo fate, i costi di storage di Postgres continueranno a crescere nel tempo. Attenzione: questo comando va eseguito all'interno di Postgres:

SELECT pg_drop_replication_slot('my_replication_slot');

Come lo facciamo in DoiT

Noi di DoiT International affrontiamo problemi come questo tutti i giorni e mi viene spesso chiesto come risparmiare durante l'implementazione dei progetti dati.

Aiutare a realizzare progetti nel modo più efficace ed economicamente vantaggioso è parte della nostra missione verso i clienti. Gestire situazioni come queste e offrire le migliori soluzioni FinOps ai clienti è ciò che facciamo — e sarò forse un po' di parte nel dirlo, ma lo facciamo MOLTO bene.

Mettiamo inoltre a disposizione strumenti di riferimento per le esigenze di ottimizzazione dei costi, come PerfectScale per Kubernetes e Select per BigQuery, Databricks e Snowflake,