S3 ClickPipe は、Amazon S3 および S3 互換オブジェクトストアから ClickHouse Cloud にデータを取り込むための、フルマネージドかつ高い耐障害性を備えた手段を提供します。一回限り と 継続的インジェスト の両方を、exactly-once セマンティクスでサポートしています。
S3 ClickPipes は、ClickPipes UI を使用して手動でデプロイおよび管理できるほか、OpenAPI や Terraform を使用してプログラムから管理することもできます。
オブジェクトストレージサービスプロバイダーごとに URL 形式や API 実装が異なるため、すべての S3 互換サービスがそのままサポートされているわけではありません。上記に記載のないサービスで問題が発生している場合は、弊社チームまでお問い合わせください。 file1、file2、file3 という名前のファイルは順番に取り込まれますが、新たに file 0 が バケット に追加されても、ファイル名が最後に取り込まれたファイルより辞書順で後ではないため、無視されます。
このモードでは、S3 ClickPipe は指定した path 内のすべてのファイルを初期ロードし、その後、設定可能な間隔 (デフォルトでは 30 秒) で新しいファイルをポーリングします。特定のファイルや時点からインジェストを開始することはできません。ClickPipes は常に、指定した path 内のすべてのファイルを読み込みます。
S3 ClickPipe は、バケットに接続された Amazon SQS キューを設定し、必要に応じてイベントルーターとして Amazon EventBridge を使用することで、暗黙的な順序を持たないファイルを取り込めるように設定できます。これにより、ClickPipes はオブジェクト作成イベントを監視し、ファイル名の規則に関係なく新しいファイルを取り込めます。
順不同モードは Amazon S3 でのみサポートされており、パブリックバケットや S3 互換サービスではサポートされません。利用するには、バケットに接続された Amazon SQS キューを設定し、必要に応じてイベントルーターとして Amazon EventBridge を使用する必要があります。 ObjectCreated:* イベントを監視します。すでに認識済みのファイルに対するメッセージ、パスに一致しないファイル、または別の種類のイベントは無視されます。
イベントにプレフィックス/サフィックスを設定するかどうかは任意です。設定する場合は、ClickPipe に設定したパスと一致していることを確認してください。S3 では、同じイベントタイプに対して重複する複数の通知ルールは許可されません。
max insert bytes または max file count で設定された閾値に達した時点、または設定可能な間隔 (デフォルトでは 30 秒) の経過後に取り込まれます。特定のファイルまたは時点からインジェストを開始することはできません。ClickPipes は常に選択したパス内のすべてのファイルをロードします。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) 。
例:
https://bucket.s3.amazonaws.com/folder/*.csvhttps://bucket.s3.amazonaws.com/logs/**/data.jsonhttps://bucket.s3.amazonaws.com/file-?.parquethttps://bucket.s3.amazonaws.com/data-2024-*.csv.gz
例:
https://bucket.s3.amazonaws.com/{documents-01,documents-02}.jsonhttps://bucket.s3.amazonaws.com/file-{1..100}.csvhttps://bucket.s3.amazonaws.com/{logs,metrics}/data.parquet
大規模なデータセットを取り込む際には、さまざまな障害が発生する可能性があり、その結果、データの一部だけが挿入されたり、重複データが発生したりすることがあります。Object Storage 用 ClickPipes は挿入失敗に対して耐性があり、exactly-once セマンティクスを提供します。これは、一時的な「ステージングテーブル」を使用して実現されます。データはまずステージングテーブルに挿入されます。この挿入中に問題が発生した場合は、ステージングテーブルを TRUNCATE し、クリーンな状態から挿入を再試行できます。挿入が正常に完了した場合にのみ、ステージングテーブル内のパーティションがターゲットテーブルへ移動されます。この戦略の詳細については、こちらのブログ記事をご覧ください。
どのファイルが取り込まれたかを追跡するには、_file 仮想カラムをカラムマッピングのリストに含めます。_file 仮想カラムにはソースオブジェクトのファイル名が含まれており、どのファイルが処理されたかをクエリで確認できます。
S3 ClickPipe は、パブリックバケットとプライベートバケットに対応しています。Requester Pays バケットは サポートされていません。
バケットポリシーで、次のアクションを許可する必要があります。
順不同モードを使用する場合、SQS のキューポリシーで次のアクションを許可する必要があります。
アクセスキーを使用して認証するには、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 を取得するには、ターミナルを開いて次を実行します。
-
VPC エンドポイントベースのアクセス制御では、S3 バケットは ClickHouse Cloud サービスと同じリージョンに配置されている必要があり、
GetObject オペレーションは ClickHouse Cloud サービスの VPC Endpoint ID に制限する必要があります。ご利用の ClickHouse Cloud リージョンの VPC エンドポイントを取得するには、ターミナルを開いて次を実行します。
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内のデータが欠落する可能性があります。 
