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

사전 조건: Module 00–02가 완료되어 있고, 터미널이 ClickHouse_Demos/workshops/build_workshop/app에 있어야 합니다.

Step 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, 호스트명, 일회성 postgres 비밀번호를 저장하세요. 베타 상태인 list와 get 명령은 빈 결과나 FORBIDDEN을 반환할 수 있습니다. 필수 준비 상태 점검은 Step 2의 ./preflight.sh --require-postgres입니다.

비밀번호를 잃어버렸다면 새로 만드세요:

clickhousectl cloud postgres reset-password <postgres-id>

Step 2 — 운행 데이터 writer 시작

로컬 Postgres 대체 수단은 없습니다. .env.workshop의 비어 있는 PGHOST와 PGPASSWORD 필드를 clickhousectl이 반환한 값으로 채우고, TLS는 required로 유지하세요:

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

매니지드 엔드포인트를 검증한 다음, 로컬 writer를 그 엔드포인트에 대해 시작하고 로그를 따라가세요:

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를 누르세요. 컨테이너는 계속 실행됩니다.

Step 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에는 이 명령을 위한 대화형 비밀번호 입력이 없습니다. 값을 임시 변수로 읽어들이면 셸 히스토리에 남지 않으며, 생성된 비밀번호가 -로 시작할 때도 --password=... 형태가 동작합니다. 명령이 실행되는 동안에는 로컬 프로세스 검사 도구에 값이 잠시 노출될 수 있으므로, 신뢰할 수 있는 머신을 사용하고 위와 같이 즉시 unset하세요.

CLI로 만든 ClickPipe는 대상을 default.realtime_trips에 둡니다. 상태를 확인하세요:

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

파이프가 실행 중이고 초기 스냅샷이 대상 테이블을 만들었으면 계속하세요.

Step 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 스킬로 설계를 점검하세요:

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.

Step 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로 설정하세요.

완료 확인

writer 로그가 반복적인 insert를 보여줍니다.
clickhousectl cloud clickpipe get ...이 실행 중인 파이프를 보고합니다.
default.realtime_trips와 대시보드 행 카운트가 모두 증가합니다.
Ops 대시보드가 수동 새로고침 없이 갱신됩니다.

04 ClickHouse Agents로 계속하세요.

결과물