Migrando seu banco de dados para um data lake Iceberg (Parte 2)

A série

Esta é a continuação da série sobre replicar seu banco de dados para um data lake baseado em Iceberg. Na Parte 1, vimos a teoria, o como e o porquê disso tudo. Nesta segunda parte, vou mostrar a implementação em si, com código, comandos e lições aprendidas ao colocar isso em produção.

Pré-requisitos da implementação

Os únicos pré-requisitos de fato são ter uma instância na nuvem (ou uma solução on-premises) que permita subir um cluster Kubernetes, uma fonte compatível com o Debezium Server e um BLOB store como GCS ou S3 para armazenar os dados das tabelas Iceberg. Neste artigo, vou dar as instruções para configurar tudo isso — menos a fonte — no GCP.

Você nem vai precisar de muito conhecimento em Kubernetes para implementar este exemplo básico, já que vou passar todos os comandos prontos. Acredite: além deste exemplo básico, você vai precisar de conhecimento sim para rodar isso em escala de produção, mas, para o que faremos aqui, não precisa ser um Kubestronaut.

Vale lembrar que dá para implantar isso de outras formas além do Kubernetes e, se for rodar isso sozinho em um cluster Kubernetes dedicado, eu recomendaria essa alternativa por questão de custo. Se você só vai rodar essa aplicação e não precisa de um cluster Kubernetes completo, considere subi-la como um container standalone em uma plataforma de computação como GCE ou EC2.

O plano de implementação

O plano desta implementação segue estes passos:

1. Preparar sua fonte (Postgres, neste caso)
2. Configurar as permissões no seu bucket/destino
3. Ajustar o deployment do Kubernetes com as suas configurações
4. Fazer o deployment no Kubernetes
5. Testar

É um plano bem simples de propósito, e vou destacando pontos em que dá para melhorar a solução para uso em produção, indo além deste caso de uso simples, que pretendo apresentar como prova de conceito.

Como costumam me chamar de "The BigQuery Dude", vou implementar isso no GCP, porque é a minha praia. Vale destacar que este exemplo é trivial de implantar também na AWS ou no Azure. Vou implantar o Debezium Server no GKE, usando o Postgres no Cloud SQL como fonte e tabelas Iceberg no GCS — tudo com equivalentes na AWS e no Azure.

Estou usando o máximo possível de código agnóstico de plataforma, mas parte dele é específica do GCP. Tudo, exceto os comandos "gcloud" e os prefixos "gs://" abaixo, é transportável diretamente para qualquer variação de Kubernetes, storage e Postgres que possa existir nos outros provedores.

Como dados de exemplo ao longo deste post, estou usando o famoso banco Northwind, com o qual muita gente da minha geração aprendeu SQL. Você pode baixar o código no GitHub aqui ou ir direto ao arquivo SQL aqui. Ele fica no schema public e estou rodando tudo com um usuário chamado "my_user" para facilitar a busca e substituição, além de conseguir executar os comandos diretamente como teste.

A implementação (replicação do banco de dados de origem)

O primeiro passo da implementação é deixar seu banco de dados pronto. Não vou entrar em detalhes de como configurar isso, porque já foi escrito inúmeras vezes. Vou supor que você está usando PostgreSQL no Cloud SQL do GCP para este caso de teste. Se você não estiver usando PostgreSQL, veja aqui como fazer esses passos no banco de dados da sua preferência.

Primeiro, garanta que a flag wal_level esteja definida como logical — isso vale principalmente para instâncias self-hosted e fora do Cloud SQL. No Cloud SQL, isso é tratado pela camada gerenciada e é chamado de flag "logical_decoding". Para adicioná-la, edite sua instância e, em flags, garanta que cloudsql.logical_decoding esteja definida como On; se ainda não tiver sido adicionada, basta adicionar e definir o valor como On.

O próximo passo é configurar direitinho o usuário de replicação. Eu recomendo FORTEMENTE criar um novo usuário para isso, com um nome descritivo, para conseguir monitorá-lo melhor. Normalmente vejo esse usuário chamado de debezium ou cdc, mas você pode escolher o nome que preferir.

Para criar um novo usuário, execute estes comandos (e veja logo abaixo o que fazer se aparecer erro):

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

Se você receber erros ao executar o comando acima dizendo que não é possível adicionar a role REPLICATION a um usuário, execute este comando no seu usuário (supondo que você tenha os privilégios; caso contrário, talvez precise pedir a um admin ou usar o usuário postgres):

ALTER USER my_user WITH REPLICATION;

Agora vem o momento da verdade: escolher qual tabela você quer replicar. Aqui estou usando uma única tabela, por simplicidade. Execute o próximo comando, especificando as tabelas e schemas e concedendo acesso ao usuário selecionado. Você pode especificar quantas quiser, separadas por vírgula:

-- Altere "my_publication" para o nome de sua preferência
CREATE PUBLICATION my_publication
FOR TABLE public.orders;

-- Execute isto para cada schema selecionado acima
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 uma aplicação mais prática do que a que estou cobrindo neste artigo, esta é a sintaxe do SQL para várias tabelas ou todas as tabelas de um schema:

-- Para várias tabelas
CREATE PUBLICATION my_publication
FOR TABLE public.orders, public.order_details;

-- Para todas as tabelas de um schema
CREATE PUBLICATION my_publication
FOR TABLES IN SCHEMA public;

-- Execute isto para cada schema selecionado acima
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 fim, precisamos criar um replication slot. Um replication slot nada mais é do que um processo que roda continuamente, publicando os dados alterados para os receptores. Estou chamando o meu de my_replication_slot, para facilitar a execução e a busca/substituição depois. Para criá-lo, execute o seguinte comando:

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

A esta altura, a configuração da replicação no banco de dados de origem está pronta (supondo que você esteja usando Postgres).

Aviso importante: se você não estiver usando esse replication slot ativamente, ele vai continuar rodando e acumulando dados alterados em disco. Ou seja, se não for usar, apague-o, porque ele vai começar a consumir seu espaço em disco ao longo do tempo, deixando sua instância de banco de dados mais cara.

Notas de segurança sobre a service account

Aqui estou rodando o cluster GKE — e, portanto, o serviço Debezium Server — como a service account padrão do Compute Engine. Faço isso apenas para fins de demonstração, para não ter que entrar no Princípio do Menor Privilégio. NÃO faça isso em produção com a service account padrão. Crie uma nova service account com apenas as permissões necessárias; um mergulho profundo nesse assunto foge do escopo deste artigo. Eu recomendo começar essa jornada por este artigo na documentação do Google.

Você também vai notar que, mais abaixo, aponto os lugares em que estou fazendo algo por simplicidade, e não por segurança, já que se trata de uma demonstração. Preste atenção nesses pontos: estou poupando você de dores de cabeça de segurança lá na frente, então use um método mais seguro. POR FAVOR!

A implementação (montando suas variáveis de ambiente)

Estou incluindo este passo para facilitar sua vida — na primeira vez em que escrevi este texto, eu não tinha feito isso, e, ao revalidar tudo antes de publicar, acrescentei este passo para deixar as coisas MUITO mais fáceis. A ideia é simplificar os comandos abaixo, evitando que você tenha que editar os valores em cada etapa. Não modifique nada após a linha tracejada, pois esses valores são gerados automaticamente.

Note que há credenciais aqui, então, se você estiver fazendo isso em produção, use um método mais seguro, como um secret manager.

O último bloco de comandos vai ecoar cada valor para você validar, exceto o valor da senha. Se aparecer algum erro, confira seus valores e certifique-se de tê-los definido corretamente.

Segue o comando, com um comentário acima de cada linha explicando cada coisa:

# Nome do projeto
PROJECT_NAME=<project_name>
# Nome do bucket onde as tabelas Iceberg vão ficar
BUCKET_NAME=<bucket-name>
# Caminho base no bucket acima onde as tabelas Iceberg vão ficar,
# normalmente um nome de tabela ou schema.
# Se estiver em dúvida, deixe vazio e ele gravará na raiz do
# bucket.
BUCKET_PATH=<bucket_path>
# Nome do seu cluster, estou usando cdc_cluster nos comandos acima
CLUSTER_NAME=<cluster_name>
# Região do cluster, estou usando us-central1 nos comandos acima.
# Esta deve ser a mesma região da sua instância Cloud SQL
CLUSTER_REGION=<cluster_region>
# Nome da instância Cloud SQL
SQL_INSTANCE_NAME=<cloud_sql_instance_name>
# Nome de usuário do banco que você criou acima
DATABASE_USER=<database_user>
# Senha do banco para o usuário que você criou acima
DATABASE_PASSWORD=<database_password>
# Nome do banco onde os comandos acima foram aplicados
DATABASE_NAME=<database_name>

#----------------------------------------------------------------------
# NÃO modifique abaixo desta linha, pois estes valores buscam valores
# adicionais e populam variáveis de ambiente geradas 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)'

# O trecho abaixo ecoa tudo o que foi definido acima, para conferência
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"

A implementação (permissões do bucket)

Para ler e gravar em um bucket do GCS, primeiro precisamos criar o bucket. Depois, temos um conjunto de permissões que precisa ser concedido à service account que executa o cluster GKE que vamos implementar na próxima seção.

Eu testei e tentei enxugar os requisitos de IAM ao máximo, comparado ao que o Google recomenda, mas, nos meus testes, as duas roles predefinidas a seguir são as melhores opções para conceder a essa service account:

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

O Google menciona isso aqui, para a implementação do BigLake, e recomendo ler essa página se você for usar BigLake. O detalhe que eles omitem é que você vai precisar das permissões da role Storage Object Admin para gravar tabelas Iceberg. Aplique essa role e pronto — não caia na armadilha de ficar tentando combinações de permissões e caçando erros. Confia em mim.

Execute o comando abaixo para criar o bucket e aplicar os bindings de política IAM da service account, usando a padrão do Compute Engine; ajuste conforme necessário:

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"

Os comandos são simples de entender. O primeiro apenas aponta o gcloud para o projeto correto. O segundo cria o bucket. Já os dois últimos concedem as permissões necessárias à service account padrão do compute do projeto.

A implementação (criando um cluster Kubernetes)

Esta implementação vai criar um cluster Kubernetes usando o Google Kubernetes Engine (GKE) Autopilot e carregar o Debezium Server nele por meio de um Helm chart. Se você já conhece o GKE ou já tem outro cluster rodando, provavelmente sabe o que está fazendo, então pode pular para os comandos de carregamento logo abaixo.

Primeiro, precisamos adicionar uma role à nossa service account padrão do Compute Engine. Aqui está o comando:

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

Em seguida, vamos criar o cluster GKE Autopilot em si. Basta rodar o comando abaixo.

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

Isso vai levar um tempinho para subir, mas, depois de alguns minutos, você deve receber a confirmação de conclusão e conseguir ver seu cluster na seção GKE do Console do GCP.

Nota sobre redes para aplicações do mundo real

Ao rodar isso em ambiente real, há uma boa chance de você estar rodando o Cloud SQL em outra região ou lidando com comunicação entre projetos. Minha maior recomendação é fazer o possível para usar o IP privado e o proxy do Cloud SQL, contornando algumas das "particularidades" das redes no GCP.

A implementação (deployment no cluster Kubernetes)

Agora vem a parte divertida: implantar o Debezium Operator no GKE. Basta rodar este comando e dar alguns minutos para ele baixar e carregar:

# Autentica o gcloud com o cluster GKE, permitindo rodar kubectl depois
gcloud container clusters get-credentials $CLUSTER_NAME --region $CLUSTER_REGION

# Adiciona o repositório Helm e atualiza a lista local de fontes Helm
helm repo add debezium https://charts.debezium.io
helm repo update

# Instala o operator em um namespace dedicado (ex.: debezium-system)
# Ele criará esse namespace caso ainda não exista
helm install debezium-operator debezium/debezium-operator \
  --namespace debezium-system \
  --create-namespace

Observação: deixei os comentários no comando explicando o que cada linha faz, o que vai ajudar você a entender o que está acontecendo e a diagnosticar possíveis problemas.

Para validar se a implantação deu certo e/ou diagnosticar por que o deploy falhou no pior cenário, execute o comando abaixo, que lista todos os pods do namespace criado e também exibe os logs associados:

# Lista os pods no namespace debezium-system
kubectl get debeziumserver -n debezium-system
# Exibe os logs dos pods
kubectl logs -l app.kubernetes.io/name=debezium-server -n debezium-system

Agora sim, você está pronto para fazer o deploy de verdade!

O comando abaixo é meio longo, mas faz uma operação relativamente simples: apenas implanta o Debezium Server e sua configuração, com um proxy do Cloud SQL anexado. Para quem tem mais familiaridade com Kubernetes, ele está implantando o Cloud SQL Proxy como sidecar em um pod que roda o Debezium Server.

Quando estiver tudo pronto, é só rodar 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
  # --- Configuração do 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
  # --- Configuração do connector ---
  source:
    connector.class: io.debezium.connector.postgresql.PostgresConnector
    config:
      # Conecta ao localhost porque o proxy está no mesmo 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"
      # Obrigatório para Cloud SQL Postgres
      plugin.name: "pgoutput"
  # --- Configuração do sink Iceberg ---
  sink:
    type: iceberg
    config:
      # Tipo do Iceberg Catalog (Hadoop é o melhor para setups só com GCS)
      debezium.sink.iceberg.catalog.type: hadoop
      debezium.sink.iceberg.catalog.warehouse: gs://$BUCKET_NAME/$BUCKET_PATH

      # Configuração do FileSystem do GCS
      debezium.sink.iceberg.fs.defaultFS: gs://$BUCKET_NAME
      debezium.sink.iceberg.io-impl: org.apache.iceberg.gcp.gcs.GCSFileIO

      # Comportamento da tabela: upsert (mescla mudanças) ou append
      debezium.sink.iceberg.upsert: "true"
      debezium.sink.iceberg.upsert.keep-deletes: "false"

      # Intervalo de commit (frequência com que um novo snapshot Iceberg é criado)
      debezium.sink.iceberg.table.commit.interval-ms: "60000"
EOF

Testando o resultado (finalmente!)

Agora é hora do teste! Entre no seu Postgres e simplesmente insira linhas na tabela contemplada pelo comando que rodamos antes. É só isso que você precisa fazer do lado do Postgres para disparar a replicação daqui para frente. Ele deve começar a gravar os dados direto em uma tabela Iceberg no GCS.

Para testar, vamos usar o BigQuery para consultar a tabela Iceberg. No BigQuery, vamos criar uma BigLake External Table. Traduzindo para quem não usa BigQuery: você está dizendo ao BigQuery que tem tabelas em um formato específico armazenadas em um bucket do GCS e quer consultá-las pelo BigQuery por meio de uma camada de abstração chamada BigLake, responsável pelas leituras no GCS.

Para isso, abra o Console do GCP e vá até a seção BigQuery, que deve abrir uma janela de editor de texto no que eles chamam de BigQuery Studio.

Cole este comando lá e ajuste com a região onde seu bucket está, o nome do bucket e o caminho do bucket que você definiu acima:

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

Execute isso; será criado um dataset específico para este teste. Assim, conseguimos remover tudo depois sem dor de cabeça.

Depois que a consulta rodar, clique na aba de nova consulta no topo do editor do BigQuery Studio. Em seguida, cole este SQL e execute:

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

Isso deve retornar algumas linhas dos dados que você inseriu. Note que ele faz um table sample de 15%, então não retornará todos os dados. Como estamos falando de BigQuery e da sua incrível capacidade de gerar contas altas, tomo essa precaução para que as consultas fiquem bem pequenas.

A esta altura, você implementou com sucesso a replicação de dados do Postgres para o Iceberg! O próximo passo é fazer o mesmo para várias tabelas (ou schemas), colocando todos os seus dados em um datastore "agnóstico de banco".

Concluindo

Este é apenas um exemplo básico, gravando uma única tabela em uma única tabela Iceberg, mas o potencial disso é imenso. Imagine conseguir gravar bancos inteiros de vendas com transações em tempo real em um data warehouse, que depois pode ser consultado para mostrar dados de vendas atualizados em dashboards e relatórios.

A melhor parte é que o data warehouse que faz a consulta não faz diferença: se você prefere BigQuery, ClickHouse, Snowflake, Databricks etc., ou uma combinação, está tudo certo. Manter-se agnóstico de plataforma é uma das melhores práticas que você pode adotar hoje para continuar competitivo, tanto tecnicamente quanto para ficar à frente nos custos.

Este é só um dos muitos temas que abordamos e sobre os quais assessoramos na DoiT International. Se tiver interesse em saber mais, fale com a gente para ver como podemos ajudar com seus custos de nuvem ou com suas operações de nuvem.

Limpeza (se/quando quiser)

Por fim, se depois de testar você quiser remover tudo para reduzir custos e planejar como rodar isso em produção, este é o comando para "botar fogo" no cluster GKE:

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

Para se livrar do dataset do BigQuery que criamos, execute isto no BigQuery Studio:

DROP SCHEMA IF EXISTS `testing_biglake_dataset`;

NÃO esqueça este passo se estiver removendo tudo, porque é ele que remove o replication slot da sua instância Postgres. Se você não fizer isso, terá custos crescentes de armazenamento no Postgres ao longo do tempo. Este comando precisa ser executado dentro do Postgres:

SELECT pg_drop_replication_slot('my_replication_slot');

Como fazemos na DoiT

Aqui na DoiT International, lidamos com problemas como esse o tempo todo, e vivem me perguntando sobre formas de economizar dinheiro na hora de implementar projetos de dados.

Ajudar a implementar projetos da forma mais eficaz e econômica possível faz parte da nossa missão com os clientes. Cuidar de tudo, de casos como este até entregar as melhores soluções de FinOps para nossos clientes, é o que fazemos — e posso estar sendo um pouco suspeito ao dizer isso, mas fazemos MUITO bem feito.

Além disso, oferecemos ferramentas best-in-class para apoiar suas necessidades de otimização de custos, como PerfectScale para Kubernetes e Select para BigQuery, Databricks e Snowflake,