Amazon S3 と ClickHouse Cloud の統合

オブジェクトストレージを ClickHouse Cloud にシームレスに接続できます。

S3 ClickPipe は、Amazon S3 および S3 互換オブジェクトストアから ClickHouse Cloud にデータを取り込むための、フルマネージドかつ高い耐障害性を備えた手段を提供します。一回限り と 継続的インジェスト の両方を、exactly-once セマンティクスでサポートしています。

S3 ClickPipes は、ClickPipes UI を使用して手動でデプロイおよび管理できるほか、OpenAPI や Terraform を使用してプログラムから管理することもできます。

​

サポート対象のデータソース

名前	ロゴ	詳細
Amazon S3		継続的インジェストでは、デフォルトで 辞書式順序 が必要ですが、任意の順序でファイルを取り込む ように設定することもできます。
Cloudflare R2 S3 互換		継続的インジェストでは 辞書式順序 が必要です。順不同モードはサポートされていません。
DigitalOcean Spaces S3 互換		継続的インジェストでは 辞書式順序 が必要です。順不同モードはサポートされていません。
OVH Object Storage S3 互換		継続的インジェストでは 辞書式順序 が必要です。順不同モードはサポートされていません。

オブジェクトストレージサービスプロバイダーごとに URL 形式や API 実装が異なるため、すべての S3 互換サービスがそのままサポートされているわけではありません。上記に記載のないサービスで問題が発生している場合は、弊社チームまでお問い合わせください。

​

対応フォーマット

• JSON
• CSV
• TSV
• Parquet
• Avro

​

機能

​

一回限りのインジェスト

デフォルトでは、S3 ClickPipe は、指定したバケット内でパターンに一致するすべてのファイルを、1 回のバッチ処理で ClickHouse の宛先テーブルに読み込みます。インジェスト タスクが完了すると、ClickPipe は自動的に停止します。この一回限りのインジェスト モードでは exactly-once セマンティクスが提供されるため、各ファイルは重複なく確実に処理されます。

​

継続的インジェスト

継続的インジェストが有効な場合、ClickPipes は指定されたパスからデータを継続的に取り込みます。インジェスト順を決める際、S3 ClickPipe はデフォルトでファイルの暗黙の辞書式順序を使用します。また、バケットに接続された Amazon SQS キューを使って、ファイルを任意の順序で取り込むように設定することもできます。

​

辞書式順序

デフォルトでは、S3 ClickPipe はファイルが バケット に辞書式順序で追加されることを前提としており、この暗黙の順序に基づいてファイルを順次取り込みます。つまり、新しいファイルは、最後に取り込まれたファイルよりも辞書順で後にある必要があります。たとえば、file1、file2、file3 という名前のファイルは順番に取り込まれますが、新たに file 0 が バケット に追加されても、ファイル名が最後に取り込まれたファイルより辞書順で後ではないため、無視されます。

このモードでは、S3 ClickPipe は指定した path 内のすべてのファイルを初期ロードし、その後、設定可能な間隔 (デフォルトでは 30 秒) で新しいファイルをポーリングします。特定のファイルの後から継続的インジェストを開始するには、Start after を設定します。ClickPipes は、設定した値以下の辞書式順序のファイルをスキップし、その時点からより新しいファイルの取り込みを続行します。

​

任意の順序

手順については、継続的インジェスト向けの順不同モードの設定を参照してください。

S3 ClickPipe は、バケットに接続された Amazon SQS キューを設定し、必要に応じてイベントルーターとして Amazon EventBridge を使用することで、暗黙的な順序を持たないファイルを取り込めるように設定できます。これにより、ClickPipes はオブジェクト作成イベントを監視し、ファイル名の規則に関係なく新しいファイルを取り込めます。

順不同モードは Amazon S3 でのみサポートされており、パブリックバケットや S3 互換サービスではサポートされません。利用するには、バケットに接続された Amazon SQS キューを設定し、必要に応じてイベントルーターとして Amazon EventBridge を使用する必要があります。

このモードでは、S3 ClickPipe は選択したパス内のすべてのファイルを初期ロードし、その後、指定したパスに一致するキュー内の ObjectCreated:* イベントを監視します。すでに認識済みのファイルに対するメッセージ、パスに一致しないファイル、または別の種類のイベントは無視されます。キュー経由で配信されたファイルのみを取り込むには、Skip initial load を有効にします。有効にすると、ClickPipes は選択したパス内の既存ファイルの初期スキャンをスキップし、SQS キュー経由で配信されたファイルのみを処理します。

イベントにプレフィックス/サフィックスを設定するかどうかは任意です。設定する場合は、ClickPipe に設定したパスと一致していることを確認してください。S3 では、同じイベントタイプに対して重複する複数の通知ルールは許可されません。

ファイルは、max insert bytes または max file count で設定された閾値に達した時点、または設定可能な間隔 (デフォルトでは 30 秒) の経過後に取り込まれます。DLQ が設定されている場合、失敗したメッセージは再度エンキューされ、DLQ の maxReceiveCount パラメータで設定された回数まで再処理されます。

失敗したメッセージのデバッグや再試行をしやすくするため、SQS キューには Dead-Letter-Queue (DLQ) を設定することを強く推奨します。

EventBridge から SQS へ

S3 イベント通知を Amazon EventBridge 経由で SQS に送信することもできます。EventBridge は、より高度なイベントフィルタリングや複数ターゲットへのファンアウトをサポートしており、S3 の「各プレフィックス・各イベントタイプにつき通知ルールは 1 つまで」という制限も受けないため、ほとんどのユースケースで推奨される方法です。手順については、継続的インジェストの順不同モードの設定 を参照してください。

SNS から SQS へ

S3 イベント通知は、SNS トピックを介して SQS に送信することもできます。これは、S3 → SQS の直接的なインテグレーションの制限に達した場合に利用できます。この場合は、raw message delivery オプションを有効にする必要があります。

​

ファイルパターンマッチング

Object Storage 用 ClickPipes では、ファイルパターンマッチングに POSIX 標準を使用します。すべてのパターンは大文字と小文字を区別し、バケット名の後ろにあるフルパス全体に対して照合されます。パフォーマンス向上のため、できるだけ具体的なパターンを使用してください (例: *.csv ではなく data-2024-*.csv) 。

​

対応しているパターン

パターン	説明	例	一致するパス
?	ちょうど 1 文字に一致します (/ を除く)	data-?.csv	data-1.csv, data-a.csv, data-x.csv
*	0 文字以上に一致します (/ を除く)	data-*.csv	data-1.csv, data-001.csv, data-report.csv, data-.csv
** 再帰	0 文字以上に一致します (/ を含む) 。ディレクトリを再帰的に走査できます。	logs/**/error.log	logs/error.log, logs/2024/error.log, logs/2024/01/error.log

例:

• https://bucket.s3.amazonaws.com/folder/*.csv
• https://bucket.s3.amazonaws.com/logs/**/data.json
• https://bucket.s3.amazonaws.com/file-?.parquet
• https://bucket.s3.amazonaws.com/data-2024-*.csv.gz

​

サポートされていないパターン

パターン	説明	例	代替手段
{abc,def}	ブレース展開	{logs,data}/file.csv	パスごとに個別の ClickPipes を作成してください。
{N..M}	数値範囲の展開	file-{1..100}.csv	file-*.csv または file-?.csv を使用してください。

例:

• https://bucket.s3.amazonaws.com/{documents-01,documents-02}.json
• https://bucket.s3.amazonaws.com/file-{1..100}.csv
• https://bucket.s3.amazonaws.com/{logs,metrics}/data.parquet

​

exactly-once セマンティクス

大規模なデータセットを取り込む際には、さまざまな障害が発生する可能性があり、その結果、データの一部だけが挿入されたり、重複データが発生したりすることがあります。Object Storage 用 ClickPipes は挿入失敗に対して耐性があり、exactly-once セマンティクスを提供します。これは、一時的な「ステージングテーブル」を使用して実現されます。データはまずステージングテーブルに挿入されます。この挿入中に問題が発生した場合は、ステージングテーブルを TRUNCATE し、クリーンな状態から挿入を再試行できます。挿入が正常に完了した場合にのみ、ステージングテーブル内のパーティションがターゲットテーブルへ移動されます。この戦略の詳細については、こちらのブログ記事をご覧ください。

​

仮想カラム

どのファイルが取り込まれたかを追跡するには、_file 仮想カラムをカラムマッピングのリストに含めます。_file 仮想カラムにはソースオブジェクトのファイル名が含まれており、どのファイルが処理されたかをクエリで確認できます。

​

アクセス制御

​

権限

S3 ClickPipe は、パブリックバケットとプライベートバケットに対応しています。Requester Pays バケットは サポートされていません。

​

S3 バケット

バケットポリシーで、次のアクションを許可する必要があります。

• s3:GetObject
• s3:ListBucket

​

SQS キュー

順不同モードを使用する場合、SQS のキューポリシーで次のアクションを許可する必要があります。

• sqs:ReceiveMessage
• sqs:DeleteMessage
• sqs:GetQueueAttributes
• sqs:ListQueues

​

認証

​

IAM 認証情報

アクセスキーを使用して認証するには、ClickPipe 接続の設定時に Authentication method で Credentials を選択します。次に、アクセスキー ID (例: AKIAIOSFODNN7EXAMPLE) とシークレットアクセスキー (例: wJalrXUtnFEMI/K7MDENG/bPxRfiCYEXAMPLEKEY) を、それぞれ Access key と Secret key に入力します。

​

IAM role

ロールベースのアクセス制御を使用して認証するには、ClickPipe 接続の設定時に Authentication method で IAM role を選択します。

S3 へのアクセスに必要な信頼ポリシーを持つロールを作成するには、このガイドに従ってください。次に、IAM role ARN に IAM role の ARN を入力します。

​

ネットワークアクセス

S3 ClickPipes では、メタデータの検出とデータのインジェストに、それぞれ ClickPipes サービスと ClickHouse Cloud サービスという 2 つの異なるネットワーク経路を使用します。追加のネットワークセキュリティ層を設定する場合 (たとえばコンプライアンス上の理由など) 、両方の経路に対してネットワークアクセスを設定する必要があります。

• IP ベースのアクセス制御では、S3 バケットポリシーで、こちらに記載されている ClickPipes サービスのリージョンの静的 IP と、ClickHouse Cloud サービスの静的 IP の両方を許可する必要があります。ご利用の ClickHouse Cloud リージョンの静的 IP を取得するには、ターミナルを開いて次を実行します。

  # <your-region> をご利用の ClickHouse Cloud リージョンに置き換えます
  curl -s https://api.clickhouse.cloud/static-ips.json | jq -r '.aws[] | select(.region == "<your-region>") | .egress_ips[]'
  

• VPC エンドポイントベースのアクセス制御では、S3 バケットは ClickHouse Cloud サービスと同じリージョンに配置されている必要があり、GetObject オペレーションは ClickHouse Cloud サービスの VPC Endpoint ID に制限する必要があります。ご利用の ClickHouse Cloud リージョンの VPC エンドポイントを取得するには、ターミナルを開いて次を実行します。

  # <your-region> をご利用の ClickHouse Cloud リージョンに置き換えます
  curl -s https://api.clickhouse.cloud/static-ips.json | jq -r '.aws[] | select(.region == "<your-region>") | .s3_endpoints[]'
  

​

高度な設定

ClickPipes には、ほとんどのユースケースの要件を満たす適切なデフォルト設定が用意されています。さらに細かい調整が必要な場合は、以下の設定を変更できます。

設定	デフォルト値	説明
Max insert bytes	10GB	1 回の挿入バッチで処理するバイト数。
Max file count	100	1 回の挿入バッチで処理するファイルの最大数。
Max threads	auto(3)	ファイル処理に使用する同時実行スレッドの最大数。
Max insert threads	1	ファイル処理に使用する同時実行の挿入スレッドの最大数。
Min insert block size bytes	1GB	テーブルに挿入できるブロックの最小バイトサイズ。
Max download threads	4	同時実行ダウンロードスレッドの最大数。
Object storage polling interval	30s	ClickHouse クラスターにデータを挿入するまでの最大待機時間を設定します。
Start after	None	順序付き継続的インジェストの場合、設定したファイル名またはパス以下の辞書順のファイルをスキップします。
Skip initial load	false	順不同継続的インジェストの場合、既存ファイルの初期スキャンをスキップし、設定した SQS キューからのオブジェクト通知のみを処理します。
Parallel distributed insert select	2	Parallel distributed insert select 設定。
Parallel view processing	false	アタッチされたビューへのプッシュを順次ではなく並列で有効にするかどうか。
Use cluster function	true	複数のノードにまたがってファイルを並列処理するかどうか。

​

スケーリング

Object Storage 用 ClickPipes は、設定済みの垂直オートスケーリング設定によって決まる最小の ClickHouse service サイズに基づいてスケールされます。ClickPipe のサイズは、パイプの作成時に決定されます。以降に ClickHouse service の設定を変更しても、ClickPipe のサイズには影響しません。

大規模な取り込みジョブのスループットを向上させるには、ClickPipe を作成する前に ClickHouse service をスケールしておくことを推奨します。

​

既知の制約事項

​

ファイルサイズ

ClickPipes が取り込みを試みるのは、サイズが10GB以下のオブジェクトのみです。ファイルが 10GB を超える場合は、ClickPipes 専用のエラーテーブルにエラーが追記されます。

​

互換性

S3 互換であっても、一部のサービスでは、S3 ClickPipe で解釈できない URL 形式が使われていたり (例: Backblaze B2) 、継続的で順不同のインジェストのためにプロバイダー固有のキューサービスとのインテグレーションが必要になったりします。サポートされているデータソースに記載のないサービスで問題が発生している場合は、当社チームまでお問い合わせください。

​

ビューのサポート

ターゲットテーブル上のmaterialized viewもサポートされています。ClickPipesは、ターゲットテーブルだけでなく、それに依存するすべてのmaterialized viewに対してもステージングテーブルを作成します。

non-materialized viewに対してはステージングテーブルを作成しません。つまり、1つ以上の下流のmaterialized viewを持つターゲットテーブルがある場合、それらのmaterialized viewでは、ターゲットテーブルのデータをview経由で選択しないようにしてください。そうしないと、materialized view内のデータが欠落する可能性があります。