Skip to content
ClickHouse Docs
ClickHouse DocsClickHouse Docs
49.6kGet Started

Pub/Sub から ClickHouse への Dataflow テンプレート

View as MarkdownEdit this pageReport an issueSet up docs MCP

Pub/Sub to ClickHouse テンプレートは、Pub/Sub サブスクリプションから JSON エンコードされたメッセージを読み取り、ClickHouse テーブルに書き込むストリーミング パイプラインです。 パースに失敗したメッセージ、またはターゲット スキーマにマッピングできなかったメッセージは、デッドレターの宛先 (ClickHouse テーブル、Pub/Sub トピック、またはその両方) にルーティングされます。

パイプラインの要件

  • ソースの Pub/Sub サブスクリプションが存在している必要があります。
  • サブスクリプションにパブリッシュされるメッセージは、有効な JSON である必要があります。
  • ClickHouse のターゲットテーブルが存在している必要があり、そのカラム名は JSON ペイロード内のフィールド名と一致している必要があります。
  • ClickHouse ホストは、Dataflow の worker マシンからアクセス可能である必要があります。
  • 少なくとも 1 つのデッドレター宛先 (clickHouseDeadLetterTable または deadLetterTopic) を指定する必要があります。両方を指定した場合、処理に失敗したメッセージは両方の宛先に同時にルーティングされます。
  • clickHouseDeadLetterTable を設定する場合、デッドレターテーブルは デッドレター処理 に示されているスキーマで、ClickHouse 内にあらかじめ存在している必要があります。
  • deadLetterTopic を設定する場合、Pub/Sub トピックはあらかじめ存在している必要があります。

Template パラメータ



Parameter Name Parameter Description Required Notes
inputSubscription メッセージの読み取り元となる Pub/Sub サブスクリプション。例: projects/<PROJECT_ID>/subscriptions/<SUBSCRIPTION_NAME>。 ✅ メッセージは JSON エンコードされている必要があります。
clickHouseUrl ClickHouse のエンドポイント URL。SSL 接続 (ClickHouse Cloud) には https://、非 SSL 接続には http:// を使用します。例: https://<HOST>:8443 または http://<HOST>:8123。 ✅ ClickHouse Cloud では、ポート 8443 の HTTPS エンドポイントを使用します。
clickHouseDatabase ターゲットテーブルが存在する ClickHouse データベースの名前。例: default。 ✅
clickHouseTable データの書き込み先となる ClickHouse テーブルの名前。 ✅ パイプラインを実行する前に、テーブルが存在している必要があります。
clickHouseUsername ClickHouse での認証に使用するユーザー名。 ✅
clickHousePassword ClickHouse での認証に使用するパスワード。 ✅
clickHouseDeadLetterTable 失敗したメッセージの書き込み先となる ClickHouse テーブル。例: my_table_dead_letter。 clickHouseDeadLetterTable または deadLetterTopic の少なくとも一方を指定する必要があります。テーブルは、デッドレター処理 に示されているデッドレターのスキーマで、あらかじめ作成されている必要があります。
deadLetterTopic 失敗したメッセージの公開先となる Pub/Sub トピック。例: projects/<PROJECT_ID>/topics/<TOPIC_NAME>。 clickHouseDeadLetterTable または deadLetterTopic の少なくとも一方を指定する必要があります。失敗したペイロードは、errorMessage と failedAt をメッセージ属性として設定してトピックに公開されます。
windowSeconds 時間ベースのバッチ処理ウィンドウの継続時間 (秒) 。 batchRowCount との相互関係については バッチ処理とウィンドウ化 を参照してください。どちらも設定されていない場合、組み合わせモードではデフォルト値の 30s と 1000 行が使用されます。
batchRowCount ClickHouse へフラッシュする前に蓄積する行数。 windowSeconds との相互関係については バッチ処理とウィンドウ化 を参照してください。
maxInsertBlockSize ClickHouse に送信する INSERT ステートメントあたりの最大行数。デフォルトは 1,000,000 です。 ClickHouseIO オプションです。
maxRetries 失敗した ClickHouse への insert に対する最大再試行回数。デフォルトは 5 です。 ClickHouseIO オプションです。
insertDeduplicate レプリケートテーブルに対する INSERT クエリで重複排除を有効にするかどうか。デフォルトは true です。 ClickHouseIO オプションです。
insertQuorum レプリケートテーブルに対する INSERT クエリで、指定した数のレプリカが書き込みを確認し、データ追加が線形化されるまで待機します。0 はクォーラム書き込みを無効にします。 ClickHouseIO オプションです。デフォルトのサーバー設定では無効です。
insertDistributedSync 有効にすると、分散テーブルへの INSERT クエリは、データがクラスター内のすべてのノードに送信されるまで待機します。デフォルトは true です。 ClickHouseIO オプションです。

メッセージ形式とスキーマのマッピング

Pub/Sub メッセージは、トップレベルのフィールド名が ClickHouse のターゲットテーブルのカラム名と完全に一致する JSON オブジェクトである必要があります。

受信メッセージをターゲットテーブルにマッピングするため、パイプラインは起動時に次の処理を実行します。

  1. ClickHouse のターゲットテーブルのスキーマを取得します。
  2. その ClickHouse スキーマから Beam の Row スキーマを構築します。
  3. 受信した各 Pub/Sub メッセージについて JSON ペイロードを解析し、ClickHouse スキーマで定義されたフィールドを読み取って行を組み立てます。

型変換

JSON 値は、対応する ClickHouse のカラム型に変換されます。

ClickHouse 型 注記
Float32 Float.valueOf で解析されます。
Float64 Double.valueOf で解析されます。
Date ISO-8601 形式の日付文字列として解析されます。
DateTime ISO-8601 形式の日時文字列 (例: 2026-01-15T12:34:56Z) として解析されます。
Array(T) JSON 配列。各要素は要素型 T に変換されます。空の配列、または存在しない配列は空配列になります。
Integer types (Int8/Int16/Int32/Int64, UInt8/UInt16/UInt32/UInt64) JSON の数値、またはその文字列表現から解析されます。
String テキストフィールドではそのまま使用されます。テキスト以外の JSON ノードは、JSON 文字列形式にシリアライズされます。

バッチ処理とウィンドウ化

このパイプラインはストリーミングで動作するため、受信した行は ClickHouse に書き出される前にウィンドウに蓄積されます。ウィンドウ化の戦略は、指定したパラメータに応じて選択されます。

windowSeconds batchRowCount 動作
設定済み 未設定 windowSeconds に基づく時間ベースの固定ウィンドウ。
未設定 設定済み 件数トリガー付きのグローバルウィンドウ。batchRowCount 行ごとにトリガーされます。
両方設定済み 両方設定済み 複合トリガー付きのグローバルウィンドウ。時間 または 行数のいずれかの条件が先に満たされた時点でトリガーされます。
どちらも未設定 どちらも未設定 既定値を使う複合モード: 30 秒または 1000 行のいずれか早い方でトリガーされます。

これらの値を調整することで、レイテンシと INSERT 効率のバランスを取れます。ウィンドウが小さいほどエンドツーエンドのレイテンシは低くなり、ウィンドウが大きいほど INSERT バッチは少なくなり、1 回あたりのサイズは大きくなります。

デッドレター処理

JSON のパース、スキーマのマッピング、または型変換に失敗したメッセージは、設定されたデッドレターの宛先にルーティングされます。clickHouseDeadLetterTable または deadLetterTopic の少なくともいずれか 1 つを指定する必要があります。両方が設定されている場合、失敗したメッセージはその両方に送信されます。

ClickHouse デッドレターテーブル

clickHouseDeadLetterTable が設定されている場合、デッドレターテーブルは次の固定スキーマですでに作成されている必要があります。

カラム 型 説明
raw_message String 元の Pub/Sub メッセージのペイロード (UTF-8 テキスト) 。
error_message String 行が失敗した理由を示す例外メッセージ。
stack_trace String 失敗時に取得された Java の完全なスタックトレース。
failed_at DateTime 行が失敗した時点の処理時刻のタイムスタンプ。

単一ノードデプロイメント向けの最小定義:

CREATE TABLE my_table_dead_letter (
    raw_message   String,
    error_message String,
    stack_trace   String,
    failed_at     DateTime
) ENGINE = MergeTree()
ORDER BY failed_at;

Pub/Sub デッドレタートピック

deadLetterTopic が設定されている場合、失敗した各メッセージは次の内容でそのトピックに再公開されます。

  • ペイロード: 元のメッセージのバイト列。
  • 属性 errorMessage: 失敗時に記録された例外メッセージ。
  • 属性 failedAt: 行の処理が失敗した時点の処理時刻タイムスタンプ。

これにより、原因となっていたスキーマやプロデューサーの問題を解消した後で、失敗したメッセージを簡単に再投入できます。

テンプレートの実行

Pub/Sub to ClickHouse テンプレートは、Google Cloud Console から利用できます。

Google Cloud Console にサインインし、Dataflow を検索します。

  1. CREATE JOB FROM TEMPLATE ボタンをクリックします。

    Dataflow コンソール

  2. テンプレートのフォームが開いたら、ジョブ名を入力し、使用するリージョンを選択します。

  3. Dataflow Template 入力欄に ClickHouse または Pub/Sub と入力し、Pub/Sub to ClickHouse テンプレートを選択します。

  4. 選択すると、フォームが展開されます。以下を入力します。

    • projects/<PROJECT_ID>/subscriptions/<SUBSCRIPTION_NAME> の形式で Pub/Sub の入力サブスクリプション。
    • ClickHouse のエンドポイント URL — ClickHouse Cloud の場合は https://<HOST>:8443 を使用します。
    • ClickHouse のデータベース、ターゲットテーブル、ユーザー名、パスワード。
    • 少なくとも 1 つの デッドレターの宛先: ClickHouse テーブルまたは Pub/Sub トピック (あるいはその両方) 。

  5. 必要に応じて、Template パラメータ セクションで詳しく説明しているとおり、バッチ処理 (windowSeconds、batchRowCount) および ClickHouseIO のチューニングパラメータをカスタマイズします。

ジョブを監視する

ジョブのステータスを監視するには、Google Cloud Console の Dataflow Jobs タブ に移動します。ここでは、進行状況やエラーを含むジョブの詳細を確認できます。

実行中の Pub/Sub to ClickHouse ジョブが表示された Dataflow コンソール

このテンプレートは、PubSubToClickHouse ネームスペース配下に次のカスタムメトリクスも出力します。これらは Dataflow のジョブページで確認できます。

Metric 種類 説明
messages-received カウンター パース処理ステップで受信した Pub/Sub メッセージの総数。
rows-parsed-ok カウンター 正常に行へ変換され、メイン出力にルーティングされたメッセージ。
rows-parse-failed カウンター パースまたはスキーママッピングに失敗し、デッドレターにルーティングされたメッセージ。
message-payload-bytes 分布 受信した Pub/Sub メッセージのpayloadサイズの分布 (バイト単位) 。

トラブルシューティング

メモリ制限 (合計) 超過エラー (コード 241)

このエラーは、大きなバッチのデータを処理している際に、ClickHouse のメモリが不足すると発生します。この問題を解決するには、次の対応を行ってください。

  • インスタンスのリソースを増やす: データ処理の負荷に対応できるよう、より多くのメモリを備えた大きなインスタンスに ClickHouseサーバーをアップグレードします。
  • バッチサイズを小さくする: Dataflow ジョブの設定で batchRowCount (および/または maxInsertBlockSize) を減らし、ClickHouse に送信するデータの chunk を小さくして、バッチごとのメモリ消費を抑えます。

すべてのメッセージが デッドレター 宛先に送られる

最も一般的な原因は次のとおりです。

  • JSON フィールド名が ClickHouse のカラム名と完全に一致していない (この照合では大文字と小文字が区別されます) 。
  • JSON の値をカラム型に変換できない (たとえば、DateTime カラムに ISO-8601 形式ではない文字列が入っている場合) 。
  • パイプラインの起動後にターゲットテーブルのスキーマが変更された — スキーマは起動時に一度だけ取得されます。スキーマ変更を適用したら、ジョブを再起動してください。

根本原因を特定するには、ClickHouse の デッドレター テーブルの error_message および stack_trace カラム (または Pub/Sub の デッドレター メッセージの errorMessage 属性) を確認してください。

パイプラインは開始するが、ClickHouse に行が届かない

  • サブスクリプションがメッセージを受信していることを確認してください。Dataflow のジョブページで messages-received メトリックを確認します。
  • 時間ベースのモード (windowSeconds のみ) では、行がフラッシュされるのはウィンドウ境界のタイミングだけです。フラッシュが発生していることを確認するため、windowSeconds を小さくしてください。
  • Dataflow ワーカーと ClickHouse エンドポイント間のネットワーク到達性を確認してください (ファイアウォール、VPC ピアリング、または Private Service Connect) 。

Template のソースコード

Template のソースコードは以下で公開されています。

Navigation