Faire évoluer votre base de données vers un data lake Iceberg (Partie 2)

La série

Cet article s'inscrit dans la continuité de notre travail sur la réplication d'une base de données vers un data lake basé sur Iceberg. Dans la Partie 1, nous avons abordé la théorie, le comment et le pourquoi. Dans cette seconde partie, je couvre l'implémentation concrète, avec le code, les commandes et les enseignements tirés de sa mise en production.

Les prérequis à l'implémentation

Les seuls véritables prérequis sont de disposer d'une instance cloud (ou d'une solution on-prem) permettant de lancer un cluster Kubernetes, d'une source compatible avec Debezium Server, et d'un stockage BLOB tel que GCS ou S3 pour héberger les données de vos tables Iceberg. Dans cet article, je vais détailler la mise en place de tout cela sur GCP, sauf pour la source.

Vous n'aurez même pas besoin de connaissances poussées en Kubernetes pour mettre en œuvre cet exemple de base, car je vous fournirai toutes les commandes à exécuter. Croyez-moi, elles seront indispensables au-delà de ce cas simple pour passer à l'échelle en production, mais pour cet exemple, pas besoin d'être un Kubestronaut.

Notez que ceci peut être déployé autrement que sur Kubernetes. Si vous exécutez cela seul sur un cluster Kubernetes, je recommande cette approche par souci d'économie. Si vous ne faites tourner que cette application et n'avez pas besoin d'un cluster Kubernetes complet, envisagez plutôt de la lancer comme conteneur autonome sur une plateforme de compute telle que GCE ou EC2.

Le plan d'implémentation

Le plan suit ces étapes :

1. Préparer votre source (Postgres dans ce cas)
2. Configurer les permissions sur votre bucket/destination
3. Adapter le déploiement Kubernetes à vos paramètres
4. Déployer sur Kubernetes
5. Tester le tout

Un plan volontairement simple. J'insérerai au fil du texte des remarques signalant les points d'amélioration pour un usage en production, au-delà de ce cas d'usage minimal que je conçois comme une preuve de concept.

Comme on m'appelle souvent "The BigQuery Dude", je vais réaliser tout cela sur GCP, qui est mon terrain de jeu naturel. Notez que cet exemple se déploie tout aussi facilement sur AWS ou Azure. Je vais déployer Debezium Server sur GKE, avec Postgres sur Cloud SQL comme source, et des tables Iceberg sur GCS — autant de services qui ont leurs équivalents chez AWS et Azure.

J'utilise autant que possible du code agnostique de la plateforme, mais certaines parties resteront spécifiques à GCP. Tout ce qui n'est pas commande "gcloud" ou préfixe "gs://" ci-dessous se transposera directement à n'importe quelle variante de Kubernetes, de stockage ou de Postgres disponible chez les autres fournisseurs.

Pour les données d'exemple tout au long de cet article, j'utilise la fameuse base de données Northwind, sur laquelle beaucoup de ma génération a appris le SQL. Vous pouvez télécharger le code depuis GitHub ici ou accéder directement au fichier SQL ici. Elle se trouve dans le schéma public, et j'exécute tout avec un utilisateur nommé "my_user" pour faciliter les rechercher-remplacer et pouvoir lancer ces commandes directement en test.

L'implémentation (réplication de la base source)

La première étape consiste à préparer votre base de données. Je n'entre pas dans les détails de cette configuration, car elle a déjà été documentée à maintes reprises. Je pars du principe que vous utilisez PostgreSQL sur Cloud SQL de GCP pour ce cas de test. Si vous n'utilisez pas PostgreSQL, consultez cette page pour connaître ces étapes avec votre base de données préférée.

D'abord, vérifiez que votre flag wal_level est réglé sur logical, notamment pour les instances auto-hébergées et non-Cloud SQL. Sur Cloud SQL, cela est géré par la couche managée et s'appelle le flag "logical_decoding". Pour l'ajouter, éditez votre instance et, sous les flags, assurez-vous que cloudsql.logical_decoding est réglé sur On ; s'il n'est pas présent, ajoutez-le simplement en veillant à ce que sa valeur soit sur On.

Ensuite, il faut configurer correctement l'utilisateur de réplication. Je recommande FORTEMENT de créer un nouvel utilisateur avec un nom explicite, afin de mieux le surveiller. On voit généralement cet utilisateur nommé debezium ou cdc, mais vous pouvez choisir ce que vous voulez.

Pour créer un nouvel utilisateur, exécutez ces commandes (et regardez plus bas en cas d'erreur) :

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

Si vous rencontrez des erreurs indiquant l'impossibilité d'attribuer le rôle REPLICATION à un utilisateur, exécutez cette commande sur votre utilisateur (en supposant que vous disposiez des privilèges ; sinon, il vous faudra un admin ou l'utilisateur postgres) :

ALTER USER my_user WITH REPLICATION;

Voici le moment de vérité : choisir la table à répliquer. Je le fais sur une seule table par souci de simplicité. Exécutez ensuite cette commande en précisant les tables et schémas, et en accordant l'accès à l'utilisateur sélectionné. Vous pouvez en indiquer autant que nécessaire, séparés par une virgule :

-- Change "my_publication" to your preferred name
CREATE PUBLICATION my_publication
FOR TABLE public.orders;

-- Run these for each schema selected above
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;

Pour un usage plus concret, au-delà de ce que je couvre ici, voici la syntaxe SQL permettant de traiter plusieurs tables, ou l'ensemble des tables d'un schéma :

-- For multiple tables
CREATE PUBLICATION my_publication
FOR TABLE public.orders, public.order_details;

-- For all tables in a schema
CREATE PUBLICATION my_publication
FOR TABLES IN SCHEMA public;

-- Run these for each schema selected above
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;

Enfin, il faut créer un replication slot. Un replication slot n'est qu'un processus qui tourne en continu et publie les données modifiées vers d'éventuels récepteurs. Je l'appelle my_replication_slot pour faciliter l'exécution et les rechercher-remplacer par la suite. Exécutez la commande suivante :

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

À ce stade, la configuration de la réplication côté base source est terminée (en supposant que vous suivez le guide avec Postgres).

Note importante : si vous n'utilisez pas activement ce replication slot, il continuera de tourner et de stocker sur disque les données modifiées. Donc, si vous ne vous en servez pas, supprimez-le, car il finira par consommer votre espace disque au fil du temps, ce qui rendra votre instance de base de données plus coûteuse.

Notes de sécurité sur le compte de service

Dans cet exemple, j'exécute le cluster GKE — et donc le service Debezium Server — avec le compte de service Compute Engine par défaut. C'est uniquement à des fins de démonstration, pour ne pas avoir à entrer dans le principe du moindre privilège. NE FAITES PAS cela en production avec le compte de service par défaut. Créez un nouveau compte de service ne disposant que des permissions nécessaires ; approfondir ce sujet dépasse le cadre de cet article. Je recommande de commencer à explorer cette piste avec l'article correspondant dans la documentation Google.

Vous remarquerez également que je signale plus bas les endroits où je privilégie la simplicité à la sécurité, puisqu'il s'agit d'une démonstration. Prenez-en note, car je vous évite ainsi bien des maux de tête liés à la sécurité par la suite, et adoptez une méthode plus sûre. S'IL VOUS PLAÎT !

L'implémentation (préparer vos variables d'environnement)

J'ajoute cette étape pour vous simplifier la vie : je ne l'avais pas prévue lors de ma première rédaction, mais en revalidant tout avant publication, je l'ai intégrée pour rendre les choses BEAUCOUP plus simples. Le but est d'alléger les commandes ci-dessous en évitant de modifier vos valeurs à chaque étape. Ne modifiez rien après la ligne pointillée, car ces valeurs sont générées.

Notez qu'il y a des identifiants ici : si vous faites cela en production, utilisez plutôt une méthode plus sécurisée, comme un gestionnaire de secrets.

Le dernier bloc de commandes affiche chaque valeur pour validation, à l'exception du mot de passe. Si vous constatez des erreurs, vérifiez vos valeurs et assurez-vous qu'elles sont correctement définies.

Voici la commande, avec un commentaire au-dessus de chaque valeur pour vous en indiquer le rôle :

# The project's name
PROJECT_NAME=<project_name>
# The bucket where the Iceberg tables will live's name
BUCKET_NAME=<bucket-name>
# Base path in the above bucket where the Iceberg tables will live,
# this is normally a table or schema name.
# If unsure just default it to empty and it will write to the root of
# the bucket.
BUCKET_PATH=<bucket_path>
# Name of your cluster, I am using cdc_cluster in the commands above
CLUSTER_NAME=<cluster_name>
# Cluster region, I am using us-central1 in the commands above.
# This should be the same region as your Cloud SQL instance
CLUSTER_REGION=<cluster_region>
# Cloud SQL instance name
SQL_INSTANCE_NAME=<cloud_sql_instance_name>
# Database username that you created above
DATABASE_USER=<database_user>
# Database password for this user you created above
DATABASE_PASSWORD=<database_password>
# Database name from where above commands were applied
DATABASE_NAME=<database_name>

#----------------------------------------------------------------------
# Do NOT modify below this line as these values are pulling additional
# values and populating generated environment variables for usage later.
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)'

# Below will echo out everything from above to make sure it looks correct
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'implémentation (permissions du bucket)

Pour lire et écrire dans un bucket GCS, il faut d'abord créer le bucket. Il faut ensuite accorder un ensemble de permissions au compte de service qui exécutera le cluster GKE mis en place dans la section suivante.

J'ai testé et cherché à simplifier les exigences IAM par rapport aux recommandations de Google, mais d'après mes tests, les deux rôles prédéfinis suivants sont les meilleures options à accorder à ce compte de service :

• Storage Object Admin (c.-à-d. roles/storage.objectAdmin)
• Storage Legacy Bucket Reader (c.-à-d. roles/storage.legacyBucketReader)

Google en parle dans le contexte de son implémentation BigLake, et je vous conseille de lire cette page si vous comptez utiliser BigLake. Le détail qu'ils omettent, c'est que vous aurez besoin des permissions du rôle Storage Object Admin pour écrire des tables Iceberg. Appliquez simplement ce rôle et ne vous lancez pas dans une chasse aux erreurs à essayer diverses combinaisons de permissions, croyez-moi.

Exécutez la commande suivante pour créer le bucket et appliquer les liaisons de politique IAM au compte de service, en utilisant celui par défaut de Compute Engine — à adapter si nécessaire :

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"

Ces commandes sont simples à comprendre. La première configure gcloud sur le bon projet. La deuxième crée le bucket. Les deux dernières accordent les permissions nécessaires au compte de service Compute par défaut du projet.

L'implémentation (création d'un cluster Kubernetes)

Cette implémentation consiste à créer un cluster Kubernetes avec Google Kubernetes Engine (GKE) Autopilot et à y charger Debezium Server via un chart Helm. Si vous connaissez GKE ou disposez déjà d'un autre cluster en fonctionnement, vous savez sans doute ce que vous faites : vous pouvez passer directement aux commandes de chargement ci-dessous.

D'abord, il faut ajouter un rôle nécessaire à notre compte de service Compute Engine par défaut. Voici la commande :

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

Ensuite, on crée le cluster GKE Autopilot proprement dit. Il suffit d'exécuter la commande ci-dessous.

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

Son démarrage prend un peu de temps, mais après quelques minutes, vous devriez recevoir la confirmation que l'opération est terminée, et vous pourrez voir votre cluster dans la section GKE de la console GCP.

Note sur le réseau pour les applications en conditions réelles

En conditions réelles, il y a de fortes chances que vous exécutiez Cloud SQL dans une autre région ou que vous ayez à gérer des communications entre projets. Ma recommandation principale : faites de votre mieux pour utiliser l'IP privée et le proxy Cloud SQL afin de contourner certaines subtilités du réseau sur GCP.

L'implémentation (déploiement sur le cluster Kubernetes)

Passons à la partie amusante : déployer l'opérateur Debezium sur GKE. Il suffit d'exécuter cette commande et de patienter quelques minutes le temps du téléchargement et du démarrage :

# Authenticate gcloud with the GKE cluster, this allows running kubectl later
gcloud container clusters get-credentials $CLUSTER_NAME --region $CLUSTER_REGION

# Add the Helm repository and update the local Helm source list
helm repo add debezium https://charts.debezium.io
helm repo update

# Install the operator into a dedicated namespace (e.g., debezium-system)
# It will create this namespace if it does not already exist
helm install debezium-operator debezium/debezium-operator \
  --namespace debezium-system \
  --create-namespace

Note : j'ai laissé les commentaires dans la commande pour expliquer ce que fait chaque étape, ce qui vous aidera à comprendre le déroulement et à diagnostiquer d'éventuels problèmes.

Pour valider le déploiement ou, dans le pire des cas, diagnostiquer un échec, exécutez la commande ci-dessous : elle liste tous les pods du namespace créé et affiche les logs associés :

# Get the pods in the debezium-system namespace
kubectl get debeziumserver -n debezium-system
# Display the logs for the pods
kubectl logs -l app.kubernetes.io/name=debezium-server -n debezium-system

Vous êtes maintenant prêt à passer au déploiement proprement dit !

La commande ci-dessous est un peu longue, mais elle réalise une opération relativement simple : déployer Debezium Server et sa configuration avec un proxy Cloud SQL attaché. Pour ceux qui maîtrisent mieux Kubernetes, elle déploie le proxy Cloud SQL en tant que sidecar sur un pod qui exécute Debezium Server.

Quand vous êtes prêt, exécutez cette commande :

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:
      # Connect to localhost because the proxy is in the same 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"
      # Required for Cloud SQL Postgres
      plugin.name: "pgoutput"
  # --- Iceberg Sink Configuration ---
  sink:
    type: iceberg
    config:
      # The Iceberg Catalog type (Hadoop is best for GCS-only setups)
      debezium.sink.iceberg.catalog.type: hadoop
      debezium.sink.iceberg.catalog.warehouse: gs://$BUCKET_NAME/$BUCKET_PATH

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

      # Table behavior: upsert (merges changes) or append
      debezium.sink.iceberg.upsert: "true"
      debezium.sink.iceberg.upsert.keep-deletes: "false"

      # Commit interval (how often to create a new Iceberg snapshot)
      debezium.sink.iceberg.table.commit.interval-ms: "60000"
EOF

Tester le résultat (enfin !)

C'est l'heure du test ! À ce stade, rendez-vous dans votre Postgres et insérez simplement des lignes dans la table concernée par la commande exécutée précédemment. C'est tout ce que vous avez à faire côté Postgres pour déclencher la réplication à partir de maintenant. Les données devraient être écrites directement dans une table Iceberg sur GCS.

Pour tester, nous allons utiliser BigQuery pour interroger la table Iceberg. Dans BigQuery, nous créerons une table externe BigLake. Pour ceux qui ne sont pas familiers avec BigQuery, cela revient à dire à BigQuery : j'ai des tables dans un format spécifique stockées sur un bucket GCS, et j'aimerais les interroger via BigQuery en passant par une couche d'abstraction appelée BigLake pour les lectures GCS.

Pour cela, ouvrez votre console GCP et rendez-vous dans la section BigQuery : cela devrait ouvrir une fenêtre d'éditeur de texte dans ce qu'ils appellent BigQuery Studio.

Collez-y cette commande et modifiez-la avec la région où réside votre bucket, le nom du bucket et le chemin du bucket définis plus haut :

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

Exécutez cette requête ; elle créera un dataset dédié à ce test. Nous pourrons ainsi tout supprimer plus tard sans aucun souci.

Une fois la requête exécutée, cliquez sur l'onglet "nouvelle requête" en haut du volet éditeur de BigQuery Studio. Collez ensuite ce SQL et exécutez-le :

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

Cela devrait renvoyer quelques lignes de données que vous avez insérées. Notez qu'un échantillon de 15 % est utilisé, la requête ne renverra donc pas toutes les données. Puisqu'on parle de BigQuery et de sa formidable capacité à faire gonfler la facture, je prends cette précaution pour n'exécuter que de très petites requêtes.

À ce stade, vous avez implémenté avec succès la réplication de données de Postgres vers Iceberg ! L'étape suivante consiste à faire de même pour plusieurs tables (ou schémas) afin de rassembler toutes vos données dans un datastore "agnostique de la base de données".

En conclusion

Ce n'est qu'un exemple basique d'écriture d'une seule table vers une seule table Iceberg, mais la puissance de cette approche est immense. Imaginez pouvoir écrire en temps réel des bases de données de vente entières, remplies de transactions, vers un data warehouse, qui peut ensuite être interrogé pour refléter des données de vente à jour dans les dashboards et les rapports.

Le meilleur, c'est que le data warehouse qui exécute les requêtes n'a pas d'importance : que vous préfériez BigQuery, ClickHouse, Snowflake, Databricks, etc., ou un mélange, vous êtes paré. Rester agnostique de plateforme est l'une des meilleures pratiques à adopter dès aujourd'hui pour rester compétitif, aussi bien techniquement que sur la maîtrise des coûts.

Ce n'est là qu'un des nombreux sujets sur lesquels nous conseillons et intervenons chez DoiT International : si vous souhaitez en savoir plus, contactez-nous pour découvrir comment nous pouvons vous aider sur vos coûts cloud ou vos opérations cloud.

Nettoyage (si/quand vous le souhaitez)

Enfin, si, après ces tests, vous souhaitez tout supprimer pour réduire les coûts et planifier une exécution en production, voici la commande pour tout raser sur le cluster GKE :

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

Pour supprimer le dataset BigQuery que nous avons créé, exécutez ceci dans BigQuery Studio :

DROP SCHEMA IF EXISTS `testing_biglake_dataset`;

N'OUBLIEZ PAS cette étape si vous supprimez tout, car elle retire le replication slot de votre instance Postgres. Sans cela, vos coûts de stockage Postgres augmenteront au fil du temps. À noter : cette commande doit être exécutée à l'intérieur de Postgres :

SELECT pg_drop_replication_slot('my_replication_slot');

Comment nous faisons, chez DoiT

Chez DoiT International, nous traitons ce type de problèmes en permanence, et on me demande sans cesse comment économiser lors de l'implémentation de projets data.

Accompagner les projets de la manière la plus efficace et la plus économique possible fait partie de notre mission auprès de nos clients. De ce genre de cas jusqu'aux meilleures solutions FinOps pour nos clients, c'est notre métier — et je suis peut-être un peu partial en le disant, mais nous le faisons TRÈS bien.

En complément, nous fournissons des outils best-in-class pour répondre à vos besoins d'optimisation des coûts, tels que PerfectScale pour Kubernetes et Select pour BigQuery, Databricks et Snowflake,