AI SREClickHouse Workshops

03 マネージド Postgres CDC

clickhousectl で ClickHouse マネージドの Postgres と ClickPipe を作成し、ライブな行をタクシーのテーブルに流し込みます。

Your computer
macOS terminal: Run workshop commands in Terminal using zsh or bash.

成果

約 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、ホスト名、そして一度だけ表示される postgres のパスワードを 保存します。ベータ版の list と get コマンドは空か FORBIDDEN を返すことがあります。 必要な準備完了チェックは、ステップ 2 の ./preflight.sh --require-postgres です。

パスワードを紛失した場合は、新しく作成します。

clickhousectl cloud postgres reset-password <postgres-id>

ステップ 2 — 乗車データのライターを起動する

ローカルの 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

マネージドのエンドポイントを検証し、それに対してローカルのライターを起動して、そのログを追います。

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 には、このコマンド用の対話的なパスワード入力プロンプトがありません。値を 一時変数に読み込むことで、シェルの履歴に残らなくなります。また --password=... の形式は、 生成されたパスワードが - で始まる場合でも動作します。コマンドの実行中は、ローカルの プロセス調査ツールから値が一瞬見える状態になるので、信頼できるマシンを使い、上記のように すぐに unset してください。

CLI で作成した ClickPipe は、ターゲットを default.realtime_trips に置きます。その状態を 確認します。

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

パイプが 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 を1つ 作成します。選ぶバリアントや編集するファイルはありません。

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.

ステップ 5 — 行が動いていることを確認する

これを約15秒あけて2回実行します。

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 に設定します。

完了チェック

  • ライターのログに、繰り返される insert が表示される。
  • clickhousectl cloud clickpipe get ... がパイプの running を報告する。
  • default.realtime_trips とダッシュボード側の行数が、どちらも増える。
  • Ops ダッシュボードが手動更新なしで更新される。

04 ClickHouse Agents に進んでください。

02 ベースアプリ

ClickHouse にデータが入った状態で、動いている NYC タクシーアプリを一巡し、拡張する前にその構造を理解します。

04 ClickHouse Agents

自分のタクシーデータ上に ClickHouse Agent を作成し、ai.clickhouse.cloud で対話的に探索します。

このページの内容

成果ステップ 1 — ClickHouse Cloud でマネージド Postgres を作成するステップ 2 — 乗車データのライターを起動するステップ 3 — ClickPipe を作成するマネージド Postgres の CA 証明書をダウンロードするステップ 4 — CDC の materialized view を作成するステップ 5 — 行が動いていることを確認する完了チェック

Track your progress?

Optional. We email a link to confirm your address; progress records once you open it.

Please use your work email address, not a personal one.

Progress tracking also requires accepting the current Terms of Service in Privacy settings.

JA