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

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` オプションです。

すべての ClickHouseIO パラメータのデフォルト値は、ClickHouseIO Apache Beam Connector を参照してください。

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

Pub/Sub メッセージは、トップレベルのフィールド名が ClickHouse のターゲットテーブルのカラム名と完全に一致する JSON オブジェクトである必要があります。受信メッセージをターゲットテーブルにマッピングするため、パイプラインは起動時に次の処理を実行します。

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

JSON フィールド名は、ClickHouse のカラム名と完全に一致している必要があります (大文字と小文字は区別されます) 。メッセージ内のフィールドのうち、ClickHouse のカラムに対応しないものは無視されます。ClickHouse のカラムに対応するフィールドが JSON ペイロード内に存在しない場合、パイプラインはそのカラムに NULL を書き込もうとします。これは、そのカラムが Nullable として宣言されている場合にのみ成功します。解析に失敗したメッセージ、値をカラムの型に変換できないメッセージ、または Nullable でないカラムに NULL を書き込もうとするメッセージは、デッドレター宛先にルーティングされます。

型変換

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;

デプロイ環境に合わせて、エンジンとORDER BY句を調整してください。レプリケートテーブルにはReplicatedMergeTreeを使用し、分散構成ではON CLUSTERを追加するほか、必要に応じてパーティション化や有効期限 (TTL) も調整してください。

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

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

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

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

テンプレートの実行

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

テンプレートの設定要件と前提条件を十分に理解するため、このドキュメント、特に上記の各セクションを必ず確認してください。

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

CREATE JOB FROM TEMPLATE ボタンをクリックします。
テンプレートのフォームが開いたら、ジョブ名を入力し、使用するリージョンを選択します。
Dataflow Template 入力欄に ClickHouse または Pub/Sub と入力し、Pub/Sub to ClickHouse テンプレートを選択します。
選択すると、フォームが展開されます。以下を入力します。
- projects/<PROJECT_ID>/subscriptions/<SUBSCRIPTION_NAME> の形式で Pub/Sub の入力サブスクリプション。
- ClickHouse のエンドポイント URL — ClickHouse Cloud の場合は https://<HOST>:8443 を使用します。
- ClickHouse のデータベース、ターゲットテーブル、ユーザー名、パスワード。
- 少なくとも 1 つのデッドレターの宛先: ClickHouse テーブルまたは Pub/Sub トピック (あるいはその両方) 。
必要に応じて、Template パラメータセクションで詳しく説明しているとおり、バッチ処理 (windowSeconds、batchRowCount) および ClickHouseIO のチューニングパラメータをカスタマイズします。

ジョブを監視する

ジョブのステータスを監視するには、Google Cloud Console の Dataflow Jobs タブに移動します。ここでは、進行状況やエラーを含むジョブの詳細を確認できます。このテンプレートは、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 のソースコードは以下で公開されています。

GoogleCloudPlatform/DataflowTemplates — Google Cloud Platform のアップストリームリポジトリ。
ClickHouse/DataflowTemplates — ClickHouse のフォーク。

​パイプラインの要件

​Template パラメータ

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

​型変換

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

​デッドレター処理

​ClickHouse デッドレターテーブル

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

​テンプレートの実行

​ジョブを監視する

​トラブルシューティング

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

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

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

​Template のソースコード

パイプラインの要件

Template パラメータ

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

型変換

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

デッドレター処理

ClickHouse デッドレターテーブル

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

テンプレートの実行

ジョブを監視する

トラブルシューティング

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

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

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

Template のソースコード