継続的インジェスト向け順不同モードの設定

S3 ClickPipes で継続的インジェスト向けに順不同モードを設定する手順ガイド。

デフォルトでは、S3 ClickPipe は、ファイルがバケットに辞書式順序で追加されることを想定しています。Amazon SQS キューをバケットに接続し、必要に応じて Amazon EventBridge をイベントルーターとして使用することで、明確な順序を持たないファイルを取り込めるように S3 ClickPipe を設定できます。これにより、ClickPipes は ObjectCreated:* イベントを待ち受け、ファイル名の命名規則に関係なく、新しいファイルを取り込めるようになります。

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

​

仕組み

このモードでは、S3 ClickPipe は選択したパス内のすべてのファイルに対して初期ロードを実行した後、指定したパスに一致するキュー内の ObjectCreated:* イベントを監視します。すでに処理済みのファイルに関するメッセージ、パスに一致しないファイル、または別種のイベントは、すべて無視されます。ファイルは、max insert bytes または max file count で設定されたしきい値に達した時点、または設定可能な間隔 (デフォルトでは 30 秒) 経過後に取り込まれます。

キュー経由で配信されたファイルのみを取り込むには、Skip initial load を有効にします。有効にすると、ClickPipes は選択したパス内の既存ファイルの初期スキャンをスキップし、SQS キュー経由で配信されたファイルのみを処理します。

データの取り込み時にはさまざまな障害が発生する可能性があり、その結果、一部のみが挿入されたり、重複データが発生したりすることがあります。Object Storage 用 ClickPipes は挿入の失敗に強く、一時的なステージングテーブルを使用して exactly-once セマンティクスを提供します。データはまずステージングテーブルに挿入され、問題が発生した場合はステージングテーブルを空にして、クリーンな状態から挿入を再試行します。挿入が正常に完了した場合にのみ、パーティションがターゲットテーブルに移動されます。

1. Amazon SQS キューの作成

   1. AWS コンソールで Simple Queue Service > Create queue に移動します。デフォルトのまま新しい標準キューを作成します。

   ClickPipe ごとに 1 つのキュー

   各 ClickPipe には専用の SQS キューが必要です。SQS では、各メッセージは 1 つのコンシューマーにしか配信されません。2 つ以上の ClickPipe (または他のアプリケーション) が同じキューを消費すると、各イベントを受信できるのはそのうち 1 つだけになり、イベントを受信しなかったパイプでは対応するファイルが気付かれないままスキップされます。これは、ClickPipes が異なるパスパターンを使用している場合でも同様です。一致しないファイルのイベントを受信したパイプはそのイベントを破棄し、そのイベントを他のパイプが受け取ることはありません。同じバケットから複数の ClickPipes で取り込むには、ClickPipe ごとに 1 つのキューを作成し、EventBridge ルールを使用してイベントを各キューに振り分けてください。

   SQS キューに Dead-Letter-Queue (DLQ) を設定することを強く推奨します。これにより、失敗したメッセージのデバッグや再試行が容易になります。DLQ が設定されている場合、失敗したメッセージは再度エンキューされ、DLQ の maxReceiveCount パラメータで設定された回数まで再処理されます。

   2. 以下の2つのオプションのいずれかを使用して、S3バケットをSQSキューに接続します。EventBridgeはファンアウトのサポート、より柔軟なイベントフィルタリングを提供し、イベントタイプおよびプレフィックスごとに通知ルールが1つに制限されるS3の制約を受けないため、ほとんどのユースケースで推奨されます。

   EventBridge 経由

   a. S3 バケットのプロパティで Event notifications > Amazon EventBridge に移動し、EventBridge への通知送信を有効にします。Save changes をクリックします。

   b. AWS Console で Amazon EventBridge > Rules > Create rule に移動します。ルールに名前を付け (例: S3ObjectCreated) 、default event bus を選択して、Next をクリックします。Build event pattern ステップで、イベントソースとして AWS events or EventBridge partner events を選択し、次のイベントパターンを手動で入力します。このとき、<bucket-name> は使用するバケット名に置き換えてください。

   {
     "source": ["aws.s3"],
     "detail-type": ["Object Created"],
     "detail": {
       "bucket": {
         "name": ["<bucket-name>"]
       }
     }
   }
   

   必要に応じて、プレフィックスまたは接尾辞で絞り込むための object.key 条件をパターンに追加できます。その場合は、ClickPipe に設定したパスと一致していることを確認してください。

   c. Select target(s) ステップで、ターゲットタイプとして AWS service を選択し、SQS queue を選びます。前のステップで作成したキューを選択してください。EventBridge が必要な IAM ロールを自動作成できるよう、Use execution role (recommended) はチェックしたままにして、Next をクリックし、ウィザードを完了します。

   d. SQS キューのアクセスポリシーを編集し、EventBridge がメッセージを送信できるようにします。<sqs-queue-arn> と <eventbridge-rule-arn> は適切な値に置き換えてください。

   {
     "Version": "2012-10-17",
     "Id": "example-ID",
     "Statement": [
       {
         "Sid": "AllowEventBridgeToSendMessage",
         "Effect": "Allow",
         "Principal": {
           "Service": "events.amazonaws.com"
         },
         "Action": "SQS:SendMessage",
         "Resource": "<sqs-queue-arn>",
         "Condition": {
           "ArnLike": {
             "aws:SourceArn": "<eventbridge-rule-arn>"
           }
         }
       }
     ]
   }
   

   Direct S3 → SQS

   a. SQS キューのアクセスポリシーを編集し、S3 バケットがメッセージを送信できるようにします。<sqs-queue-arn>、<bucket-arn>、<aws-account-id> は適切な値に置き換えてください。

   {
     "Version": "2012-10-17",
     "Id": "example-ID",
     "Statement": [
       {
         "Sid": "AllowS3ToSendMessage",
         "Effect": "Allow",
         "Principal": {
           "Service": "s3.amazonaws.com"
         },
         "Action": "SQS:SendMessage",
         "Resource": "<sqs-queue-arn>",
         "Condition": {
           "ArnLike": {
             "aws:SourceArn": "<bucket-arn>"
           },
           "StringEquals": {
             "aws:SourceAccount": "<aws-account-id>"
           }
         }
       }
     ]
   }
   

   b. S3 バケットのプロパティで Event notifications を有効にし、イベントタイプとして All object create events (s3:ObjectCreated:*) を選択して、宛先を SQS キューに設定します。代わりに個別のイベントタイプを選択しないでください。アップロード方法によって発生する ObjectCreated イベントの種類が異なるためです (たとえば、multipart upload では Put ではなく CompleteMultipartUpload が発生します) 。また、選択していないイベントタイプに該当するファイルはキューに配信されません。必要に応じて、通知をトリガーするオブジェクトを絞り込むためにプレフィックスまたは接尾辞を指定できます。その場合は、ClickPipe に設定したパスと一致していることを確認してください。

   S3 では、同じバケット内で同じイベントタイプを対象とする、重複する複数の通知ルールは設定できません。このバケットに対してすでに ObjectCreated イベントの通知ルールがある場合は、代わりに EventBridge の方法を使用してください。

2. IAM ロールを設定する

   1. ClickHouse Cloud console で Settings > Network security information に移動し、サービスの IAM role ARN をコピーします。

   2. AWS Console で IAM > Roles > Create role に移動します。Custom trust policy を選択し、以下の内容を貼り付けます。<ch-cloud-arn> は、前の手順でコピーした IAM role ARN に置き換えてください。

   {
     "Version": "2012-10-17",
     "Statement": [
       {
         "Sid": "AllowAssumeRole",
         "Effect": "Allow",
         "Principal": {
           "AWS": "<ch-cloud-arn>"
         },
         "Action": "sts:AssumeRole"
       }
     ]
   }
   

   3. S3 のオブジェクトを読み取り、SQS キュー内のメッセージを管理するために必要な権限を含むインラインポリシーを IAM ロールに作成します。<bucket-arn> と <sqs-queue-arn> は適切な値に置き換えてください。

   {
     "Version": "2012-10-17",
     "Statement": [
       {
         "Sid": "S3BucketMetadataAccess",
         "Effect": "Allow",
         "Action": [
           "s3:GetBucketLocation",
           "s3:ListBucket"
         ],
         "Resource": "<bucket-arn>"
       },
       {
         "Sid": "AllowGetListObjects",
         "Effect": "Allow",
         "Action": [
           "s3:Get*",
           "s3:List*"
         ],
         "Resource": "<bucket-arn>/*"
       },
       {
         "Sid": "SQSNotificationsAccess",
         "Effect": "Allow",
         "Action": [
           "sqs:DeleteMessage",
           "sqs:ListQueues",
           "sqs:ReceiveMessage",
           "sqs:GetQueueAttributes"
         ],
         "Resource": "<sqs-queue-arn>"
       }
     ]
   }
   
3. 順不同モードで ClickPipe を作成する

   1. ClickHouse Cloudコンソールで Data Sources > Create ClickPipe に移動し、Amazon S3 を選択します。S3 バケットへの接続に必要な情報を入力します。Authentication method で IAM role を選択し、前の手順で作成したロールの ARN を指定します。

   2. Incoming data で Continuous ingestion をオンにします。インジェストモードとして Any order を選択し、バケットに接続されたキューの SQS queue URL を指定します。キュー内のファイルのみを処理し、選択したパスにある既存ファイルの初回スキャンをスキップするには、Skip initial load を有効にします。このキューはこの ClickPipe 専用に使用する必要があります。詳しくは Amazon SQS キューの作成 を参照してください。

   3. Parse information で、ターゲットテーブルの Sorting key を定義します。マッピングされたスキーマに必要な調整を加えたら、ClickPipes のデータベースユーザーに割り当てるロールを設定します。

   4. 設定内容を確認し、Create ClickPipe をクリックします。Skip initial load が有効でない限り、ClickPipes はまずバケットを初回スキャンして、指定したパスに一致する既存のファイルをすべて読み込みます。その後、一致する ObjectCreated:* イベントがキューから配信されると、ファイルの処理を開始します。