03 CDC avec Postgres managé

Créez Postgres managé par ClickHouse et un ClickPipe avec clickhousectl, puis alimentez la table de taxis avec des lignes en direct.

Les commandes de cette page utilisent les valeurs enregistrées dans .env.workshop.

Résultat attendu

En environ 20 minutes, les trajets en direct circuleront ainsi :

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

Prérequis : les modules 00 à 02 sont terminés et votre terminal se trouve dans ClickHouse_Demos/workshops/build_workshop/app.

Étape 1 — Créer Postgres managé dans ClickHouse Cloud

Utilisez la même région que votre service ClickHouse :

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

Enregistrez l’identifiant Postgres, le nom d’hôte et le mot de passe à usage unique de l’utilisateur postgres renvoyés. Les commandes bêta list et get peuvent renvoyer un résultat vide ou FORBIDDEN ; le contrôle de disponibilité à utiliser est ./preflight.sh --require-postgres à l’étape 2.

Si vous perdez le mot de passe, générez-en un nouveau :

clickhousectl cloud postgres reset-password <postgres-id>

Étape 2 — Démarrer le générateur de trajets

Il n’existe aucune solution de repli vers un Postgres local. Dans .env.workshop, renseignez les champs PGHOST et PGPASSWORD avec les valeurs renvoyées par clickhousectl ; conservez l’obligation d’utiliser TLS :

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

Validez le point de terminaison managé, puis démarrez le générateur local en le connectant à celui-ci et suivez son journal :

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

Le provisionnement peut prendre quelques minutes. Poursuivez lorsque le journal répète :

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

Appuyez sur Ctrl-C pour arrêter de suivre le journal ; le conteneur continue de fonctionner.

Étape 3 — Créer le ClickPipe

Remplacez l’identifiant du service ClickHouse et les valeurs Postgres :

Téléchargez le certificat d’autorité du Postgres géré

Téléchargez le certificat d’autorité spécifique à l’instance depuis Settings → Security → Download CA certificate, puis enregistrez-le sous .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 ne propose aucune saisie interactive du mot de passe pour cette commande. La lecture de la valeur dans une variable temporaire évite de la conserver dans l’historique de l’interpréteur ; la forme --password=... fonctionne également lorsque le mot de passe généré commence par -. La valeur reste brièvement visible par les outils locaux d’inspection des processus pendant l’exécution de la commande : utilisez donc une machine de confiance et supprimez aussitôt la variable, comme ci-dessus.

Un ClickPipe créé depuis la CLI place cette cible dans default.realtime_trips. Vérifiez son état :

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

Poursuivez lorsque le pipeline fonctionne et que l’instantané initial a créé la table cible.

Étape 4 — Créer la vue matérialisée de CDC

Commencez par vérifier précisément la table source :

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

Résultat attendu : default.realtime_trips. Créez maintenant une seule vue matérialisée incrémentielle ; il n’existe ni variante à choisir, ni fichier à modifier :

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
"

Utilisez la compétence de bonnes pratiques installée pour vérifier la conception :

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.

Étape 5 — Prouver que les lignes circulent

Exécutez deux fois la commande suivante à environ 15 secondes d’intervalle :

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
"

Les deux nombres doivent augmenter. Ouvrez ensuite localhost:8080, réglez l’intervalle du tableau de bord Ops sur 1m et l’actualisation automatique sur 5s.

Vérification finale

Le journal du générateur affiche des insertions répétées.
clickhousectl cloud clickpipe get ... indique un pipeline en cours d’exécution.
Le nombre de lignes dans default.realtime_trips et celui du tableau de bord augmentent tous deux.
Le tableau de bord Ops s’actualise sans intervention manuelle.

Passez à 04 ClickHouse Agents.