03 CDC con Postgres administrado

Crea un Postgres administrado por ClickHouse y un ClickPipe con clickhousectl y, a continuación, envía filas en directo a la tabla de taxis.

Los comandos de esta página usan los valores guardados en .env.workshop.

Resultado

En unos 20 minutos, los viajes en directo seguirán este flujo:

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

Requisitos previos: módulos 00 a 02 completos y terminal situada en ClickHouse_Demos/workshops/build_workshop/app.

Paso 1: crea Postgres administrado en ClickHouse Cloud

Usa la misma región que tu servicio ClickHouse:

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

Guarda el Postgres ID, el hostname y la contraseña de un solo uso del usuario postgres que se devuelven. Los comandos beta list y get pueden devolver vacío o FORBIDDEN; la comprobación de disponibilidad obligatoria es ./preflight.sh --require-postgres en el paso 2.

Si pierdes la contraseña, crea otra:

clickhousectl cloud postgres reset-password <postgres-id>

Paso 2: inicia el escritor de viajes

No hay una alternativa local de Postgres. En .env.workshop, rellena los campos vacíos PGHOST y PGPASSWORD con los valores devueltos por clickhousectl; mantén TLS como requisito:

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

Valida el endpoint administrado, inicia el escritor local contra él y sigue su registro:

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

El aprovisionamiento puede tardar varios minutos. Continúa cuando el registro repita:

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

Pulsa Ctrl-C para dejar de seguir el registro; el contenedor sigue en ejecución.

Paso 3: crea ClickPipe

Sustituye el ID del servicio ClickHouse y los valores de Postgres:

Descarga el certificado de CA de Postgres gestionado

Descarga el certificado de CA específico de la instancia en Settings → Security → Download CA certificate y guárdalo 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 no ofrece una solicitud interactiva de contraseña para este comando. Leer el valor en una variable temporal evita que permanezca en el historial del shell; el formato --password=... también funciona cuando la contraseña generada empieza por -. El valor sigue siendo visible brevemente para las herramientas locales de inspección de procesos mientras se ejecuta el comando, así que usa un equipo de confianza y elimina la variable de inmediato, como se muestra.

Un ClickPipe creado por la CLI sitúa este destino en default.realtime_trips. Comprueba su estado:

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

Continúa cuando el pipe esté en ejecución y el snapshot inicial haya creado la tabla de destino.

Paso 4: crea la vista materializada de CDC

Primero, verifica la tabla de origen exacta:

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. Ahora crea una sola vista materializada incremental; no hay ninguna variante que seleccionar ni archivo que 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
"

Usa la skill instalada de buenas prácticas para revisar el diseño:

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.

Paso 5: demuestra que las filas se mueven

Ejecuta lo siguiente dos veces, con unos 15 segundos de diferencia:

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
"

Ambos recuentos deben aumentar. Después, abre localhost:8080, configura el intervalo del panel Ops como 1m y la actualización automática como 5s.

Comprobación final

El registro del escritor muestra inserciones repetidas.
clickhousectl cloud clickpipe get ... indica que el pipe está en ejecución.
Aumentan tanto default.realtime_trips como el recuento de filas del panel.
El panel Ops se actualiza sin una recarga manual.

Continúa en 04 ClickHouse Agents.