Amazon S3 と 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 互換		継続的インジェストでは 辞書式順序 が必要です。順不同モードはサポートされていません。

• 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 はオブジェクト作成イベントを監視し、ファイル名の規則に関係なく新しいファイルを取り込めます。

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

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

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

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

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

例:

• 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

例:

• 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

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

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

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

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

• s3:GetObject
• s3:ListBucket

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

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

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

ロールベースのアクセス制御を使用して認証するには、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 には、ほとんどのユースケースの要件を満たす適切なデフォルト設定が用意されています。さらに細かい調整が必要な場合は、以下の設定を変更できます。

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内のデータが欠落する可能性があります。