Skip to content
ClickHouse Docs
ClickHouse DocsClickHouse Docs
49.6kGet Started

継続的インジェスト向けの順不同モードの設定

View as MarkdownEdit this pageReport an issueSet up docs MCP

デフォルトでは、GCS ClickPipe は、ファイルが 辞書式順序 でバケットに追加されることを前提としています。Google Cloud Pub/Sub サブスクリプションをバケットに接続するよう設定することで、暗黙的な順序を持たないファイルも取り込めるように GCS ClickPipe を構成できます。これにより、ClickPipes は OBJECT_FINALIZE 通知を受信して、ファイル名の規則に関係なく新しいファイルを取り込めるようになります。

仕組み

このモードでは、GCS ClickPipe は選択したパス内のすべてのファイルを初期ロードした後、指定したパスに一致するオブジェクト通知を Pub/Sub サブスクリプション 経由で待ち受けます。すでに確認済みのファイルに対するメッセージ、パスに一致しないファイル、または異なる種類のイベントはすべて無視されます。

サブスクリプション経由で配信されたファイルのみを取り込むには、Skip initial load を有効にします。有効にすると、ClickPipes は選択したパス内の既存ファイルに対する初期スキャンをスキップし、Pub/Sub サブスクリプション経由で配信されたファイルのみを処理します。

データの取り込み中にはさまざまな障害が発生する可能性があり、その結果、部分的な insert や重複データが生じることがあります。Object Storage 用 ClickPipes は insert の失敗に対して耐性があり、一時的なステージングテーブルを使用して exactly-once セマンティクスを実現します。データはまずステージングテーブルに挿入され、問題が発生した場合はステージングテーブルを TRUNCATE して、クリーンな状態から insert を再試行します。insert が正常に完了した場合にのみ、パーティションがターゲットテーブルに移動されます。

Google Cloud Pub/Sub topic を作成する

1. Google Cloud Console で Pub/Sub > Topics > Create topic に移動します。デフォルト サブスクリプション 付きで新しい topic を作成し、Topic Name を控えておきます。

2. 上で作成した Pub/Sub topic に OBJECT_FINALIZE events を送信する GCS bucket notification を設定します。

2.1. この手順は Google Cloud Console では実行できないため、gcloud クライアントまたは Google Cloud 用の任意のプログラムによるインターフェイスを使用する必要があります。たとえば、gcloud を使う場合は次のとおりです。

# Create a Pub/Sub notification for new objects in the bucket
gcloud storage buckets notifications create "gs://${YOUR_BUCKET_NAME}" \
  --topic="projects/${YOUR_PROJECT_ID}/topics/${YOUR_TOPIC_NAME}" \
  --event-types="OBJECT_FINALIZE" \
  --payload-format="json"

# List the Pub/Sub notifications in the bucket
gcloud storage buckets notifications describe

サービス アカウント を設定する

1. サービス アカウント を設定し、ClickPipes が指定した bucket 内のオブジェクトを一覧表示および取得できるようにするとともに、Pub/Sub サブスクリプション からの通知を受信して監視できるよう、必要な permissions を付与します。

1.1. この手順は Google Cloud Console、gcloud クライアント、または Google Cloud 用の任意のプログラムによるインターフェイスで実行できます。たとえば、gcloud を使う場合は次のとおりです。

# 1. Grant read access to the GCS bucket
gcloud storage buckets add-iam-policy-binding "gs://${YOUR_BUCKET_NAME}" \
  --member="serviceAccount:${YOUR_SERVICE_ACCOUNT}@${YOUR_PROJECT_ID}.iam.gserviceaccount.com" \
  --role="roles/storage.objectViewer"

# 2. Grant read access to the Pub/Sub subscription
gcloud pubsub subscriptions add-iam-policy-binding "${YOUR_SUBSCRIPTION_NAME}" \
  --member="serviceAccount:${YOUR_SERVICE_ACCOUNT}@${YOUR_PROJECT_ID}.iam.gserviceaccount.com" \
  --role="roles/pubsub.subscriber"

# 3. Grant permission to get the Pub/Sub subscription metadata
gcloud pubsub subscriptions add-iam-policy-binding "${YOUR_SUBSCRIPTION_NAME}" \
  --member="serviceAccount:${YOUR_SERVICE_ACCOUNT}@${YOUR_PROJECT_ID}.iam.gserviceaccount.com" \
  --role="roles/pubsub.viewer"

順不同モードで ClickPipe を作成する

1. ClickHouse Cloud console で Data Sources > Create ClickPipe に移動し、Google Cloud Storage を選択します。GCS bucket に接続するための情報を入力します。Authentication method では サービス アカウント を選択し、.json の サービス アカウント key を指定します。

2. 継続的インジェスト をオンにし、インジェストモードとして Any order を選択したうえで、bucket に接続されている サブスクリプション の Pub/Sub サブスクリプション 名を指定します。サブスクリプション のファイルのみを処理し、選択したパスにある既存ファイルの初期スキャンをスキップするには、初期ロードをスキップ を有効にします。この サブスクリプション はこの ClickPipe 専用に使用する必要があります。詳しくは Google Cloud Pub/Sub topic を作成する を参照してください。サブスクリプション 名は次の形式に従う必要があります。

projects/${YOUR_PROJECT_ID}/subscriptions/${YOUR_SUBSCRIPTION_NAME}

3. Incoming data をクリックします。ターゲットテーブルの Sorting key を定義します。マッピングされたスキーマに必要な調整を加えた後、ClickPipes データベースユーザーのロールを設定します。

4. 設定内容を確認し、Create ClickPipe をクリックします。Skip initial load が有効でない限り、ClickPipes は bucket の初期スキャンを実行して、指定したパスに一致する既存のすべてのファイルを読み込み、その後、サブスクリプションから一致する OBJECT_FINALIZE events が配信されるとファイルの処理を開始します。

Navigation