Criar uma pipeline

Este guia monta uma integração completa do zero, com as três famílias de peça:

[Input: SQL Server CDC] → tópico → [Transform: Go Function] → tópico → [Output: PostgreSQL]

O que ela faz: cada alteração feita na tabela dbo.pedidos de um SQL Server é capturada, recebe um campo fixo indicando de qual sistema veio, e é aplicada na tabela public.pedidos de um PostgreSQL.

O campo fixo é o exemplo mais simples de transformação, e serve de molde para qualquer outra: se você consegue adicionar um campo, consegue renomear, converter e filtrar do mesmo jeito.

Antes de começar

Na plataforma:

• Um broker Kafka criado em Kafka → Brokers.
• Os Secrets com a senha do SQL Server e a do PostgreSQL.

No SQL Server de origem:

• CDC habilitado no banco e na tabela dbo.pedidos, com o SQL Server Agent rodando.
• Um usuário com leitura na tabela e nas estruturas de CDC.

No PostgreSQL de destino, a tabela já criada, incluindo a coluna que o Transform vai preencher:

CREATE TABLE public.pedidos (
  id       integer PRIMARY KEY,
  cliente  text,
  total    numeric(12,2),
  origem   text                 -- preenchida pelo Transform
);

A coluna precisa existir antes

O Output do PostgreSQL descarta em silêncio qualquer campo que a tabela de destino não tenha. Sem a coluna origem, a pipeline funciona, não dá erro nenhum — e o campo simplesmente não aparece.

1. Crie a pipeline

Em Platform → Pipelines, crie uma pipeline nova e dê a ela um nome curto e descritivo. O nome aparece nos tópicos, nas métricas e nos logs, então vale escolher bem: erp-pedidos-para-postgres diz muito mais do que teste2.

2. Adicione o Input

O editor abre com a paleta de peças à esquerda, separada em Input, Transform e Output.

Arraste SQL Server CDC do grupo Input e preencha host, porta, usuário, o secret da senha, o banco, e o campo Tabelas com dbo.pedidos.

Todos os campos, com padrões e explicação, estão em Input SQL Server CDC.

Ao salvar, a plataforma cria automaticamente o tópico de saída da tabela — este Input cria um tópico por tabela configurada, não um só para todas.

3. Adicione o Transform

Arraste Go Function do grupo Transform. É aqui que o campo fixo é acrescentado.

Abra o editor de código. Ele contém o arquivo inteiro — pacote, imports e as funções — e o exemplo abaixo pode ser colado por cima:

package main

import (
	"context"
	"encoding/json"

	"go.uber.org/zap"
)

const origemFixa = "erp-vendas"

func initVars(ctx context.Context, log *zap.Logger) error {
	return nil
}

func process(value string, key string, headers map[string]string, log *zap.Logger) ([]ProcessResult, error) {
	// Exclusões chegam sem Value — a linha a apagar está na Key. Não há o que enriquecer.
	if value == "" {
		return []ProcessResult{{Value: value, Key: key, Headers: headers}}, nil
	}

	var linha map[string]any
	if err := json.Unmarshal([]byte(value), &linha); err != nil {
		return nil, err
	}

	linha["ORIGEM"] = origemFixa

	saida, err := json.Marshal(linha)
	if err != nil {
		return nil, err
	}
	return []ProcessResult{{Value: string(saida), Key: key, Headers: headers}}, nil
}

Três detalhes que fazem esse código funcionar:

Os headers são repassados intactos. O Output do PostgreSQL depende deles para saber de qual tabela a linha veio e o que fazer com ela. Devolver Headers: headers é o que preserva isso — montar headers novos na mão quebraria a gravação.

A Key é repassada intacta. É ela que identifica a linha no destino.

Exclusões são tratadas à parte. Quando uma linha é apagada na origem, a mensagem chega com o Value vazio — não há JSON para decodificar. Sem o if value == "" do começo, o json.Unmarshal falharia, a mensagem seria reprocessada para sempre e a fila travaria naquele ponto.

O campo foi escrito como ORIGEM, em maiúsculas, para acompanhar o estilo das colunas que vêm do SQL Server. O PostgreSQL compara nomes de coluna sem diferenciar maiúsculas de minúsculas, então ele grava normalmente na coluna origem.

Detalhes do contrato da função em Transform — Go Function.

4. Adicione o Output

Arraste PostgreSQL do grupo Output e preencha host, porta, database, usuário, o secret da senha, o SSL Mode e o schema padrão.

Depois monte o mapeamento de tabelas — a parte mais importante:

Campo	Valor
Schema de origem	dbo
Tabela de origem	pedidos
Tabela de destino	public.pedidos
Chave primária	id

A chave é o que permite ao Output saber se uma alteração é uma linha nova ou a atualização de uma existente, e precisa ter índice único no destino — no exemplo, a PRIMARY KEY já cumpre isso.

Todos os campos estão em Output PostgreSQL.

5. Ligue as três peças

Puxe uma ligação do Input até o Transform, e do Transform até o Output.

Nenhuma peça fala diretamente com outra: o que você está criando é a assinatura de cada peça no tópico da anterior. O caminho fica assim:

[SQL Server CDC]
      ↓  tópico da tabela dbo.pedidos
[Go Function]
      ↓  tópico de saída da função
[PostgreSQL]

Com mais de uma tabela

Se você configurar várias tabelas no Input, cada uma ganha o seu próprio tópico — e o Transform precisa ser ligado a todos eles. O Transform continua tendo um único tópico de saída, que alimenta o Output.

Um mesmo tópico também pode alimentar várias peças. Se depois você quiser mandar os mesmos pedidos enriquecidos para um bucket GCS, é só adicionar o Output e ligá-lo ao tópico do Transform — o dado não é lido duas vezes da origem.

6. Publique

Vá para a tela Deploy da pipeline. Antes do primeiro deploy, passe pela aba Variáveis e configure os valores do cluster de destino; depois, na aba Novo Deploy, escolha o cluster, a branch, as tags e escreva a mensagem do commit. O passo a passo completo está em Publicar uma pipeline.

A validação roda antes e recusa desenhos que não funcionariam — um Output sem entrada, um campo obrigatório em branco, um secret que não existe mais.

Ao subir, o Output do PostgreSQL ainda confere se a tabela, as colunas e o índice único mapeados existem de verdade. Um mapeamento errado falha na hora, com a mensagem do que está errado.

7. Teste

Primeiro, acompanhe a carga inicial: o Input copia o conteúdo atual da tabela antes de começar a capturar mudanças novas. Ela termina quando a contagem de linhas estabiliza e o lag dos tópicos cai para perto de zero.

Depois, faça uma alteração na origem:

-- No SQL Server
UPDATE dbo.pedidos SET total = 249.90 WHERE id = 42;

E confira no destino, poucos segundos depois:

-- No PostgreSQL
SELECT id, cliente, total, origem FROM public.pedidos WHERE id = 42;

 id | cliente |  total  |   origem
----+---------+---------+------------
 42 | ACME    |  249.90 | erp-vendas

O total novo veio da origem; a origem foi acrescentada pelo Transform.

Vale testar também uma exclusão, que é o caminho que mais costuma esconder problema:

DELETE FROM dbo.pedidos WHERE id = 42;

A linha deve sumir de public.pedidos.

Quando algo não chega

1. O Input capturou? Se o contador não subiu, o problema está antes da plataforma — CDC desabilitado, SQL Server Agent parado, credencial sem permissão nas estruturas de CDC.
2. O Transform deixou passar? Compare o contador de entrada com o de saída da função. Se entra e não sai, o código está devolvendo slice vazio ou dando erro — os logs do node mostram qual.
3. O Output entregou? Olhe os logs dele. Quase sempre é credencial errada, coluna que não existe ou chave sem índice único.
4. As mensagens estão acumulando no tópico? O Output está mais lento que o Input. Nada foi perdido — ele vai alcançar.
5. Chegou, mas faltando um campo? A coluna não existe no destino. O Output descarta campos que a tabela não tem, sem reclamar.

Detalhes em Observabilidade. Você também pode perguntar direto à Aruna: "a pipeline erp-pedidos-para-postgres está com problema, o que está acontecendo?"

Variações a partir daqui

Quero	Mude
Replicar sem transformar nada	Pule o passo 3 e ligue o Input direto ao Output
Mandar para data warehouse	Troque o Output por Snowflake
Receber por API em vez de banco	Troque o Input por HTTP
Descartar linhas em vez de enriquecer	Devolva nil, nil no Transform para as que não interessam