03 托管 Postgres CDC

用 clickhousectl 创建 ClickHouse 托管的 Postgres 和一个 ClickPipe，然后把实时数据行送入出租车表。

成果

在约 20 分钟 内，实时行程数据将开始流动：

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

前置条件：模块 00–02 已完成，并且你的终端位于 ClickHouse_Demos/workshops/build_workshop/app。

第 1 步：在 ClickHouse Cloud 中创建托管 Postgres

使用与你的 ClickHouse 服务相同的区域：

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

保存返回的 Postgres ID、hostname 和一次性的 postgres 密码。处于 beta 阶段的 list 和 get 命令可能返回空结果或 FORBIDDEN；必须做的就绪检查是第 2 步中的 ./preflight.sh --require-postgres。

如果密码丢失了，就创建一个新的：

clickhousectl cloud postgres reset-password <postgres-id>

第 2 步：启动行程写入器

没有本地 Postgres 的兜底方案。在 .env.workshop 中，用 clickhousectl 返回的值填写空着的 PGHOST 和 PGPASSWORD 字段；保持要求 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

先验证这个托管端点，然后启动本地写入器连上它，并跟踪它的日志：

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

资源开通可能需要几分钟。当日志开始重复出现下面的内容时再继续：

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

按 Ctrl-C 停止跟踪日志；容器会继续运行。

第 3 步：创建 ClickPipe

替换 ClickHouse 服务 ID 和 Postgres 相关的值：

下载托管 Postgres 的 CA 证书

在 Settings → Security → Download CA certificate 下载该实例专用的 CA 证书，并保存为 .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 对这条命令没有交互式的密码输入提示。把值读进一个临时变量可以避免它进入 shell 历史记录；当生成的密码以 - 开头时，--password=... 这种形式同样有效。命令运行期间该值仍会短暂地对本地的进程检查工具可见，所以请使用一台可信的机器，并按上面所示立刻 unset 它。

用 CLI 创建的 ClickPipe 会把目标放在 default.realtime_trips。检查它的状态：

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

当 pipe 处于 running 状态、并且初始快照已经创建出目标表时，再继续。

第 4 步：创建 CDC materialized view

先确认源表的确切位置：

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

预期输出：default.realtime_trips。现在创建一个增量 materialized view；这里没有变体可选，也没有文件需要编辑：

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
"

用已安装的 best-practices skill 检查这个设计：

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.

第 5 步：证明数据行正在流动

把下面这条运行两次，间隔约 15 秒：

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
"

两个计数都应该增加。然后打开 localhost:8080，把 Ops 看板的时间间隔设为 1m，自动刷新设为 5s。

完成检查

写入器日志显示反复出现的插入记录。
clickhousectl cloud clickpipe get ... 报告 pipe 处于 running 状态。
default.realtime_trips 和看板行数计数都在增加。
Ops 看板在不手动刷新的情况下自动更新。

继续前往 04 ClickHouse Agents。

成果