Iceberg Data Lake Migration: Praxis-Guide (Teil 2)

Die Serie

Dies ist die Fortsetzung zum Thema Replikation Ihrer Datenbank in einen Iceberg-basierten Data Lake. In Teil 1 ging es um Theorie, Vorgehen und Motivation. In diesem zweiten Teil widme ich mich der eigentlichen Umsetzung – inklusive Code, Befehlen und Learnings aus dem produktiven Einsatz.

Voraussetzungen für die Umsetzung

Die einzigen echten Voraussetzungen sind eine Cloud-Instanz (oder eine On-Prem-Lösung), auf der sich ein Kubernetes-Cluster hochziehen lässt, eine mit Debezium Server kompatible Quelle sowie ein BLOB-Speicher wie GCS oder S3 für die Daten Ihrer Iceberg-Tabellen. Alles außer der Quelle richte ich in diesem Artikel auf GCP ein.

Für dieses Basisbeispiel brauchen Sie nicht einmal viel Kubernetes-Wissen, denn ich liefere Ihnen sämtliche Befehle mit. Glauben Sie mir: Für den vollen produktiven Einsatz jenseits dieses Beispiels wird tieferes Wissen nötig – aber Kubestronaut müssen Sie hierfür nicht sein.

Die Anwendung lässt sich übrigens auch anders als über Kubernetes ausrollen. Wenn Sie nur diese Anwendung allein auf einem Kubernetes-Cluster betreiben, würde ich aus Kostengründen davon abraten. Läuft ausschließlich diese Anwendung und Sie brauchen keinen kompletten Kubernetes-Cluster, dann betreiben Sie sie besser als eigenständigen Container auf einer Compute-Plattform wie GCE oder EC2.

Der Umsetzungsplan

Der Plan sieht folgende Schritte vor:

Quelle vorbereiten (in diesem Fall Postgres)
Berechtigungen für Bucket/Ziel einrichten
Kubernetes-Deployment mit Ihren Einstellungen anpassen
Kubernetes-Deployment ausrollen
Testen

Ein bewusst schlichter Plan. Ich weise unterwegs an einigen Stellen darauf hin, wo sich für den produktiven Einsatz jenseits dieses Proof-of-Concept Verbesserungen anbieten.

Da ich häufig als "The BigQuery Dude" bezeichnet werde, setze ich das Ganze auf GCP um – schließlich mein Heimatterrain. Auf AWS oder Azure lässt sich dieses Beispiel genauso einfach umsetzen. Ich deploye Debezium Server auf GKE, nutze Postgres auf Cloud SQL als Quelle und Iceberg-Tabellen auf GCS – für alles davon gibt es Pendants bei AWS und Azure.

Ich schreibe den Code so plattformunabhängig wie möglich, manches bleibt aber GCP-spezifisch. Alles außer den "gcloud"-Befehlen und den "gs://"-Präfixen unten lässt sich direkt auf beliebige Kubernetes-, Storage- und Postgres-Varianten anderer Anbieter übertragen.

Als Beispieldaten nutze ich die berühmte Northwind-Datenbank, mit der viele aus meiner Generation SQL gelernt haben. Den Code finden Sie auf GitHub hier, direkt die SQL-Datei hier. Sie liegt im Schema public, und ich führe alles mit einem Benutzer namens "my_user" aus, damit Suchen und Ersetzen einfach bleibt und Sie die Befehle direkt zum Testen ausführen können.

Die Umsetzung (Replikation der Quelldatenbank)

Im ersten Schritt bereiten wir die Datenbank vor. Ich gehe hier nicht ins Detail, weil das bereits zigfach dokumentiert wurde. Ich gehe davon aus, dass Sie für diesen Testfall PostgreSQL auf GCPs Cloud SQL nutzen. Falls nicht, finden Sie hier die passenden Schritte für Ihre Datenbank.

Zunächst muss das Flag wal_level auf logical stehen – vor allem bei selbst gehosteten Instanzen außerhalb von Cloud SQL. In Cloud SQL übernimmt das die Managed-Ebene über das Flag "logical_decoding". Bearbeiten Sie dazu Ihre Instanz und stellen Sie unter Flags sicher, dass cloudsql.logical_decoding auf On steht. Falls es noch nicht existiert, fügen Sie es hinzu und setzen den Wert auf On.

Als Nächstes richten wir den Replikationsbenutzer korrekt ein. Ich empfehle DRINGEND, dafür einen eigenen Benutzer mit sprechendem Namen anzulegen, damit sich das besser überwachen lässt. Üblich sind Namen wie debezium oder cdc – Sie können aber frei wählen.

Um einen neuen Benutzer anzulegen, führen Sie folgende Befehle aus (und schauen Sie weiter unten nach, falls ein Fehler auftritt):

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

Falls beim Ausführen des obigen Befehls ein Fehler auftritt, weil die REPLICATION-Rolle nicht zugewiesen werden kann, führen Sie diesen Befehl auf Ihrem Benutzer aus (sofern Sie die nötigen Rechte haben; andernfalls braucht es einen Admin oder den postgres-Benutzer):

ALTER USER my_user WITH REPLICATION;

Jetzt kommt der Moment der Wahrheit: Welche Tabelle wollen Sie replizieren? Der Einfachheit halber nehme ich hier eine einzelne Tabelle. Führen Sie den folgenden Befehl aus, geben Sie Tabellen und Schemas an und erteilen Sie dem gewählten Benutzer Zugriff. Sie können beliebig viele Tabellen komma-separiert angeben:

-- "my_publication" ggf. durch Ihren Wunschnamen ersetzen
CREATE PUBLICATION my_publication
FOR TABLE public.orders;

-- Diese Befehle für jedes oben ausgewählte Schema ausführen
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;

Für einen praxisnäheren Einsatz jenseits dieses Artikels hier die SQL-Syntax für mehrere oder alle Tabellen eines Schemas:

-- Für mehrere Tabellen
CREATE PUBLICATION my_publication
FOR TABLE public.orders, public.order_details;

-- Für alle Tabellen in einem Schema
CREATE PUBLICATION my_publication
FOR TABLES IN SCHEMA public;

-- Diese Befehle für jedes oben ausgewählte Schema ausführen
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;

Zuletzt legen wir einen Replication Slot an. Ein Replication Slot ist schlicht ein dauerhaft laufender Prozess, der geänderte Daten an alle Empfänger publiziert. Ich nenne ihn my_replication_slot, um das spätere Ausführen und Suchen/Ersetzen zu erleichtern. Führen Sie dazu folgenden Befehl aus:

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

Damit ist die Einrichtung der Replikation in der Quelldatenbank abgeschlossen (sofern Sie mit Postgres arbeiten).

Wichtiger Hinweis: Wenn Sie diesen Replication Slot nicht aktiv nutzen, läuft er trotzdem weiter und speichert geänderte Daten auf der Festplatte. Löschen Sie ihn also, wenn Sie ihn nicht brauchen – sonst frisst er garantiert mit der Zeit Ihren Speicherplatz und macht Ihre Datenbank-Instanz teurer.

Sicherheitshinweise zum Service Account

Hier betreibe ich den GKE-Cluster und damit den Debezium-Server-Dienst unter dem Standard-Service-Account von Compute Engine. Das dient rein der Demonstration, damit wir nicht ins Principle of Least Privilege einsteigen müssen. Tun Sie das im Produktivbetrieb NICHT mit dem Standard-Service-Account. Legen Sie einen eigenen Service Account mit exakt den benötigten Rechten an – das Thema sprengt den Rahmen dieses Artikels. Als Einstieg empfehle ich den entsprechenden Artikel in der Google-Dokumentation.

Außerdem weise ich weiter unten an einigen Stellen darauf hin, wo ich es der Einfachheit halber nicht sicher mache. Nehmen Sie diese Hinweise ernst und nutzen Sie die sicherere Methode – so ersparen Sie sich später Kopfschmerzen. BITTE!

Die Umsetzung (Umgebungsvariablen aufbauen)

Diesen Schritt füge ich hinzu, um Ihnen das Leben leichter zu machen. Beim ersten Schreiben hatte ich das nicht gemacht, und bei der abschließenden Validierung vor der Veröffentlichung habe ich diesen Schritt ergänzt – er erspart Ihnen VIEL Arbeit. Die Idee: Sie müssen Ihre Werte nicht in jedem Befehl einzeln bearbeiten. Ändern Sie unterhalb der gestrichelten Linie nichts, denn diese Werte werden generiert.

Hier stehen Zugangsdaten drin – nutzen Sie im Produktivbetrieb bitte eine sichere Methode wie einen Secret Manager.

Der letzte Befehlsblock gibt jeden Wert zur Validierung aus, mit Ausnahme des Passworts. Falls Sie Fehler sehen, prüfen Sie Ihre Werte und stellen Sie sicher, dass alles korrekt gesetzt ist.

Hier der Befehl mit Kommentaren zu jeder Zeile:

# Projektname
PROJECT_NAME=<project_name>
# Name des Buckets, in dem die Iceberg-Tabellen liegen
BUCKET_NAME=<bucket-name>
# Basispfad im obigen Bucket, in dem die Iceberg-Tabellen liegen,
# üblicherweise ein Tabellen- oder Schemaname.
# Bei Unsicherheit einfach leer lassen – dann wird in den Root des
# Buckets geschrieben.
BUCKET_PATH=<bucket_path>
# Name Ihres Clusters, ich verwende cdc_cluster in den Befehlen oben
CLUSTER_NAME=<cluster_name>
# Cluster-Region, ich verwende us-central1 in den Befehlen oben.
# Sollte dieselbe Region wie Ihre Cloud-SQL-Instanz sein
CLUSTER_REGION=<cluster_region>
# Name der Cloud-SQL-Instanz
SQL_INSTANCE_NAME=<cloud_sql_instance_name>
# Datenbank-Benutzername, den Sie oben erstellt haben
DATABASE_USER=<database_user>
# Datenbank-Passwort für den oben erstellten Benutzer
DATABASE_PASSWORD=<database_password>
# Datenbankname, auf dem die obigen Befehle ausgeführt wurden
DATABASE_NAME=<database_name>

#----------------------------------------------------------------------
# UNTERHALB dieser Linie NICHT ändern – hier werden zusätzliche Werte
# ermittelt und generierte Umgebungsvariablen für die spätere Verwendung befüllt.
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)'

# Der folgende Block gibt alles Obige aus, damit Sie prüfen können, ob es passt
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"

Die Umsetzung (Bucket-Berechtigungen)

Um einen GCS-Bucket lesen und beschreiben zu können, müssen wir ihn zunächst anlegen. Danach braucht der Service Account, unter dem der GKE-Cluster läuft (siehe nächster Abschnitt), einige Berechtigungen.

Ich habe die IAM-Anforderungen getestet und versucht, sie schlanker zu gestalten als von Google empfohlen. In meinen Tests haben sich diese beiden vordefinierten Rollen als beste Wahl für diesen Service Account erwiesen:

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

Google erwähnt dies im Kontext seiner BigLake-Implementierung – die Lektüre lohnt sich, wenn Sie BigLake nutzen wollen. Ein Detail bleibt dort ungenannt: Um Iceberg-Tabellen zu schreiben, benötigen Sie die Berechtigungen der Storage-Object-Admin-Rolle. Vergeben Sie diese Rolle einfach, statt sich durch Fehler und Rechte-Kombinationen zu quälen – vertrauen Sie mir.

Führen Sie folgenden Befehl aus, um den Bucket anzulegen und die IAM-Bindings zu setzen. Ich nutze hier den Standard-Service-Account von Compute Engine – passen Sie das nach Bedarf an:

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"

Die Befehle sind leicht nachvollziehbar. Der erste stellt gcloud auf das richtige Projekt. Der zweite legt den Bucket an. Die letzten beiden vergeben die nötigen Berechtigungen an den Standard-Compute-Service-Account des Projekts.

Die Umsetzung (Kubernetes-Cluster anlegen)

Wir legen den Kubernetes-Cluster mit Google Kubernetes Engine (GKE) Autopilot an und laden den Debezium Server per Helm-Chart darauf. Wer mit GKE vertraut ist oder bereits einen Cluster betreibt, weiß vermutlich, was zu tun ist, und kann direkt zu den Befehlen weiter unten springen.

Zuerst müssen wir dem Standard-Service-Account von Compute Engine eine weitere Rolle geben. Hier der Befehl:

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

Als Nächstes erstellen wir den GKE-Autopilot-Cluster selbst. Führen Sie einfach den folgenden Befehl aus:

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

Das dauert einen Moment. Nach ein paar Minuten sollten Sie eine Bestätigung erhalten, und Ihren Cluster sehen Sie dann im Bereich GKE der GCP Console.

Hinweis zum Networking im Praxiseinsatz

In der Praxis läuft Cloud SQL oft in einer anderen Region oder es geht um projektübergreifende Kommunikation. Meine dringende Empfehlung: Nutzen Sie nach Möglichkeit die private IP und den Cloud SQL Proxy, um einige der "Eigenheiten" beim Networking auf GCP zu umgehen.

Die Umsetzung (Deployment auf dem Kubernetes-Cluster)

Jetzt kommt der spaßige Teil: das Deployment des Debezium Operator auf GKE. Führen Sie diesen Befehl aus und geben Sie ihm ein paar Minuten zum Herunterladen und Starten:

# gcloud beim GKE-Cluster authentifizieren, damit später kubectl läuft
gcloud container clusters get-credentials $CLUSTER_NAME --region $CLUSTER_REGION

# Helm-Repository hinzufügen und lokale Helm-Quellenliste aktualisieren
helm repo add debezium https://charts.debezium.io
helm repo update

# Operator in einen eigenen Namespace installieren (z. B. debezium-system)
# Der Namespace wird angelegt, falls er noch nicht existiert
helm install debezium-operator debezium/debezium-operator \
  --namespace debezium-system \
  --create-namespace

Hinweis: Ich habe die Kommentare im Befehl belassen, damit klar wird, was jeder Schritt tut – das hilft beim Verständnis und bei der Diagnose möglicher Probleme.

Um zu prüfen, ob das Deployment steht, bzw. um im Worst Case die Ursache eines fehlgeschlagenen Deployments zu finden, führen Sie den folgenden Befehl aus. Er listet alle Pods im angelegten Namespace und zeigt die zugehörigen Logs:

# Pods im Namespace debezium-system abrufen
kubectl get debeziumserver -n debezium-system
# Logs der Pods anzeigen
kubectl logs -l app.kubernetes.io/name=debezium-server -n debezium-system

Jetzt sind Sie bereit für das eigentliche Deployment!

Der folgende Befehl ist etwas länger, macht aber etwas relativ Einfaches: Er deployt Debezium Server samt Konfiguration mit einem angehängten Cloud SQL Proxy. Für Kubernetes-Kenner: Er deployt den Cloud SQL Proxy als Sidecar auf einem Pod, auf dem Debezium Server läuft.

Wenn Sie so weit sind, führen Sie diesen Befehl aus:

kubectl apply -f - <<EOF
apiVersion: debezium.io/v1alpha1
kind: DebeziumServer
metadata:
  name: debezium-sql-proxy
  namespace: debezium-system
spec:
  image: debezium/server:latest
  # --- Sidecar-Konfiguration ---
  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-Konfiguration ---
  source:
    connector.class: io.debezium.connector.postgresql.PostgresConnector
    config:
      # Verbindung zu localhost, da der Proxy im selben Pod läuft
      database.hostname: "127.0.0.1"
      database.port: "5432"
      database.user: "$DATABASE_USER"
      database.password: "$DATABASE_PASSWORD"
      database.dbname: "$DATABASE_NAME"
      topic.prefix: "cdc"
      # Erforderlich für Cloud SQL Postgres
      plugin.name: "pgoutput"
  # --- Iceberg-Sink-Konfiguration ---
  sink:
    type: iceberg
    config:
      # Iceberg-Catalog-Typ (Hadoop ist ideal für reine GCS-Setups)
      debezium.sink.iceberg.catalog.type: hadoop
      debezium.sink.iceberg.catalog.warehouse: gs://$BUCKET_NAME/$BUCKET_PATH

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

      # Tabellenverhalten: upsert (Änderungen mergen) oder append
      debezium.sink.iceberg.upsert: "true"
      debezium.sink.iceberg.upsert.keep-deletes: "false"

      # Commit-Intervall (wie oft ein neuer Iceberg-Snapshot erstellt wird)
      debezium.sink.iceberg.table.commit.interval-ms: "60000"
EOF

Ergebnis testen (endlich!)

Zeit zum Testen! Gehen Sie jetzt in Ihre Postgres-Datenbank und fügen Sie einfach Zeilen in die Tabelle ein, die von dem zuvor ausgeführten Befehl abgedeckt wird. Mehr müssen Sie auf der Postgres-Seite nicht tun, um die laufende Replikation anzustoßen. Die Daten sollten direkt in eine Iceberg-Tabelle auf GCS geschrieben werden.

Zum Testen nutzen wir BigQuery, um die Iceberg-Tabelle abzufragen. In BigQuery legen wir eine BigLake External Table an. Für alle, die BigQuery nicht kennen: Sie teilen BigQuery damit mit, dass Sie Tabellen in einem bestimmten Format haben, die in einem GCS-Bucket liegen, und dass Sie sie über eine Abstraktionsschicht namens BigLake (die die GCS-Reads übernimmt) abfragen möchten.

Öffnen Sie dazu die GCP Console und navigieren Sie zum BigQuery-Bereich. Es öffnet sich ein Texteditor-Fenster in dem Bereich, der BigQuery Studio genannt wird.

Fügen Sie folgenden Befehl ein und passen Sie Region, Bucket-Name und Bucket-Pfad an Ihre oben gesetzten Werte an:

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

Führen Sie das aus – es legt ein eigenes Dataset dafür an. So lässt sich später alles rückstandsfrei entfernen.

Klicken Sie danach oben im BigQuery-Studio-Editor auf "Neue Abfrage erstellen". Fügen Sie folgendes SQL ein und führen Sie es aus:

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

Es sollten einige der eingefügten Zeilen zurückkommen. Beachten Sie, dass ein Table Sample von 15 % gezogen wird, es kommen also nicht alle Daten zurück. Da wir hier über BigQuery reden und dessen fantastische Fähigkeit, hohe Rechnungen zu produzieren, treffe ich diese Vorsichtsmaßnahme, damit nur sehr kleine Abfragen laufen.

An diesem Punkt haben Sie erfolgreich Daten von Postgres nach Iceberg repliziert! Der nächste Schritt: dasselbe für mehrere Tabellen (oder Schemas), um all Ihre Daten in einem "datenbankagnostischen" Datastore zu haben.

Fazit

Das hier ist nur das einfache Schreiben einer einzelnen Tabelle in eine einzelne Iceberg-Tabelle – aber das Potenzial dahinter ist enorm. Stellen Sie sich vor, komplette Sales-Datenbanken mit Transaktionen in Echtzeit in ein Data Warehouse zu schreiben, das dann für Abfragen bereitsteht und aktuelle Verkaufszahlen in Dashboards und Reports liefert.

Das Beste: Welches Data Warehouse abfragt, spielt keine Rolle – ob BigQuery, ClickHouse, Snowflake, Databricks etc. oder ein Mix daraus, alles funktioniert. Plattformunabhängig zu bleiben ist eine der besten Praktiken, um heute wettbewerbsfähig zu bleiben – technisch wie kostenseitig.

Das ist nur eines von vielen Themen, zu denen wir bei DoiT International beraten. Wenn Sie mehr erfahren möchten, sprechen Sie uns an – wir zeigen Ihnen, wie wir bei Cloud-Kosten oder Cloud-Betrieb unterstützen können.

Aufräumen (falls gewünscht)

Wenn Sie nach dem Test alles wieder abbauen möchten, um Kosten zu sparen und das produktive Setup zu planen, hier der Befehl, um auf dem GKE-Cluster "alles abzureißen":

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

Um das erstellte BigQuery-Dataset loszuwerden, führen Sie das in BigQuery Studio aus:

DROP SCHEMA IF EXISTS `testing_biglake_dataset`;

Vergessen Sie diesen Schritt beim Aufräumen NICHT – er entfernt den Replication Slot aus Ihrer Postgres-Instanz. Wenn Sie ihn stehen lassen, steigen Ihre Postgres-Speicherkosten mit der Zeit. Der Befehl muss innerhalb von Postgres ausgeführt werden:

SELECT pg_drop_replication_slot('my_replication_slot');

How We DoiT

Bei DoiT International lösen wir solche Probleme laufend, und ich werde ständig gefragt, wie sich bei Datenprojekten Geld sparen lässt.

Projekte möglichst effektiv und möglichst kosteneffizient umzusetzen, ist Teil unserer Mission für unsere Kund:innen. Von Fällen wie diesem bis hin zu den besten FinOps-Lösungen für unsere Kund:innen – genau das machen wir, und ich bin vielleicht ein wenig voreingenommen, wenn ich sage: Wir machen das SEHR gut.

Darüber hinaus bieten wir erstklassige Tools für Ihre Kostenoptimierung, etwa PerfectScale für Kubernetes und Select für BigQuery, Databricks und Snowflake,