03 CDC com Postgres gerenciado

Crie um Postgres gerenciado pelo ClickHouse e um ClickPipe com o clickhousectl e, depois, envie linhas ao vivo para a tabela de táxis.

Os comandos desta página usam os valores salvos em .env.workshop.

Resultado

Em cerca de 20 minutos, as viagens ao vivo seguirão este fluxo:

Postgres managed by ClickHouse → ClickPipe → default.realtime_trips
→ materialized view → nyc_tlc_data.taxi_trips → Ops dashboard

Pré-requisitos: módulos 00 a 02 concluídos e terminal aberto em ClickHouse_Demos/workshops/build_workshop/app.

Etapa 1 — Crie o Postgres gerenciado no ClickHouse Cloud

Use a mesma região do seu serviço ClickHouse:

clickhousectl cloud postgres create \
  --name my-workshop-postgres \
  --provider aws \
  --region ap-southeast-1 \
  --size c6gd.large \
  --pg-version 17 \
  --ha-type none

Salve o Postgres ID, o hostname e a senha de uso único do usuário postgres retornados. Os comandos beta list e get podem retornar vazio ou FORBIDDEN; a verificação de prontidão obrigatória é ./preflight.sh --require-postgres, na etapa 2.

Se perder a senha, crie uma nova:

clickhousectl cloud postgres reset-password <postgres-id>

Etapa 2 — Inicie o gravador de viagens

Não há alternativa local para o Postgres. No .env.workshop, preencha os campos vazios PGHOST e PGPASSWORD com os valores retornados por clickhousectl; mantenha o TLS obrigatório:

PGHOST=replace-with-hostname-from-clickhousectl
PGPORT=5432
PGDATABASE=postgres
PGUSER=postgres
PGPASSWORD=replace-with-one-time-password
PGSSLMODE=require
PG_PUBLICATION=pub_taxi

Valide o endpoint gerenciado, inicie o gravador local apontando para ele e acompanhe seu log:

cd "$(git rev-parse --show-toplevel)/workshops/build_workshop/app"
./preflight.sh --require-postgres
docker compose --profile cdc --env-file .env.workshop -f docker-compose.workshop.yml up -d pg-trip-writer
docker compose --profile cdc --env-file .env.workshop -f docker-compose.workshop.yml logs -f pg-trip-writer

O provisionamento pode levar alguns minutos. Prossiga quando o log repetir:

[loadgen] ensured realtime_trips table exists
[loadgen] created publication pub_taxi for public.realtime_trips
[loadgen] inserted 10 trips @ ...

Pressione Ctrl-C para parar de acompanhar o log; o contêiner continuará em execução.

Etapa 3 — Crie o ClickPipe

Substitua o ID do serviço ClickHouse e os valores do Postgres:

Baixe o certificado de CA do Postgres gerenciado

Baixe o certificado da CA específico da instância em Settings → Security → Download CA certificate e salve-o como .work/managed-postgres-ca.pem.

mkdir -p .work
mv "$HOME/Downloads/<downloaded-ca-filename>" .work/managed-postgres-ca.pem

workshop_env() { sed -n "s/^$1=//p" .env.workshop | tail -n 1; }
PGHOST=$(workshop_env PGHOST)
PGPORT=$(workshop_env PGPORT)
PGDATABASE=$(workshop_env PGDATABASE)
PGUSER=$(workshop_env PGUSER)
PGPASSWORD=$(workshop_env PGPASSWORD)
PG_PUBLICATION=$(workshop_env PG_PUBLICATION)
unset -f workshop_env
PG_CA_CERT="$PWD/.work/managed-postgres-ca.pem"
test -s "$PG_CA_CERT" || { echo "Missing CA certificate: $PG_CA_CERT" >&2; exit 1; }

clickhousectl cloud clickpipe create postgres <clickhouse-service-id> \
  --name taxi-cdc \
  --host "$PGHOST" \
  --port "$PGPORT" \
  --pg-database "$PGDATABASE" \
  --username "$PGUSER" \
  --password="$PGPASSWORD" \
  --publication-name "$PG_PUBLICATION" \
  --ca-certificate "$PG_CA_CERT" \
  --replication-mode cdc \
  --table-mapping "public.realtime_trips:realtime_trips"

clickhousectl não tem um prompt interativo de senha para esse comando. A leitura do valor em uma variável temporária impede que ele fique no histórico do shell; o formato --password=... também funciona quando a senha gerada começa com -. O valor ainda fica brevemente visível para ferramentas locais de inspeção de processos enquanto o comando é executado. Portanto, use um computador confiável e apague a variável imediatamente, como mostrado.

Um ClickPipe criado pela CLI coloca o destino em default.realtime_trips. Confira seu estado:

clickhousectl cloud clickpipe list <clickhouse-service-id>
clickhousectl cloud clickpipe get <clickhouse-service-id> <clickpipe-id>

Prossiga quando o pipe estiver em execução e o snapshot inicial tiver criado a tabela de destino.

Etapa 4 — Crie a view materializada de CDC

Primeiro, verifique a tabela de origem exata:

clickhousectl cloud service query --id <clickhouse-service-id> --query "
  SELECT database, name, engine
  FROM system.tables
  WHERE name = 'realtime_trips'
"

Resultado esperado: default.realtime_trips. Agora crie uma única view materializada incremental; não há variante para selecionar nem arquivo para editar:

clickhousectl cloud service query --id <clickhouse-service-id> --query "
CREATE MATERIALIZED VIEW IF NOT EXISTS nyc_tlc_data.realtime_trips_to_taxi_trips_mv
TO nyc_tlc_data.taxi_trips
AS
SELECT
  car_type,
  CAST(vendor_id AS UInt16) AS vendor_id,
  CAST(pickup_datetime AS DateTime('UTC')) AS pickup_datetime,
  CAST(dropoff_datetime AS DateTime('UTC')) AS dropoff_datetime,
  CAST(pickup_location_id AS UInt16) AS pickup_location_id,
  CAST(dropoff_location_id AS UInt16) AS dropoff_location_id,
  CAST(passenger_count AS UInt16) AS passenger_count,
  trip_distance,
  CAST(payment_type AS UInt16) AS payment_type,
  fare_amount,
  tip_amount,
  total_amount,
  'realtime_cdc' AS filename
FROM default.realtime_trips
WHERE _peerdb_is_deleted = 0
"

Use a skill instalada de boas práticas para conferir o projeto:

Use the ClickHouse best-practices skill to review this incremental materialized view.
Confirm why it processes inserted blocks and why FINAL is not part of this append-only path.

Etapa 5 — Comprove que as linhas estão fluindo

Execute o comando abaixo duas vezes, com um intervalo de aproximadamente 15 segundos:

clickhousectl cloud service query --id <clickhouse-service-id> --query "
  SELECT
    (SELECT count() FROM default.realtime_trips) AS clickpipe_rows,
    (SELECT count() FROM nyc_tlc_data.taxi_trips
      WHERE filename = 'realtime_cdc') AS dashboard_rows
"

As duas contagens devem aumentar. Em seguida, abra localhost:8080, defina o intervalo do painel Ops como 1m e a atualização automática como 5s.

Verificação de conclusão

O log do gravador mostra inserções repetidas.
clickhousectl cloud clickpipe get ... informa que o pipe está em execução.
As contagens de default.realtime_trips e das linhas do painel aumentam.
O painel Ops é atualizado sem uma recarga manual.

Continue em 04 ClickHouse Agents.