S3 ClickPipes の Unordered モードの提供を開始

データ統合パイプラインの受信側に立つと、現実のデータには何のルールもないことをすぐに思い知らされます。データが届くこともあれば届かないこともあり、2 回届くこともあれば順不同で届くこともあります。特にオブジェクトストレージでは、これが興味深い課題を生み出します。常に大量のオブジェクトをスキャンすることなく、新しいデータをいかに効率よく追跡するかという問題です。

世にある大半のツールと同様に、ClickPipes もこれまでこの課題に対する優れたソリューションを備えておらず、ファイルが辞書順でバケットに配置されるよう「何らかの」ルールを担保することをユーザー側に求めていました。それも、もう過去の話です！

S3 ClickPipe に新しいモードを追加し、Amazon SQS 経由の S3 イベント通知を利用して任意の順序でファイルを取り込めるようにしました。バックフィル、リトライ、遅延到着データ、その他どのような原因で順不同が発生していても、適切に処理できます。

なぜこれが重要なのか？ #

デフォルトでは、S3 ClickPipe はファイルが辞書順 (lexicographical order) でバケットに追加されることを前提としており、この暗黙の順序に依存してファイルを順次取り込みます。言葉として耳慣れないだけでなく、辞書順ということは、新しく追加されるファイルが最後に取り込まれたファイルよりも辞書順で「後」でなければならないことを意味します。たとえば、events_2024-12-01.parquet、events_2024-12-02.parquet、events_2024-12-03.parquet という名前のファイルは順番通りに取り込まれますが、後から events_2024-11-30.parquet というバックフィルファイルがバケットに追加されても、そのファイルは無視されてしまいます。

これは決して望ましい動作ではありません。

順不同モード (unordered mode) を使えば、この制約はなくなります。次に処理すべきファイルを探すためにバケットを 30 秒ごとにポーリングする代わりに、ClickPipe は新しいファイルが追加されたという通知を待つようになります。通知が届くと、ClickPipes は過去に処理されたファイルとの相対的な順序に関係なく、そのファイルをそのまま処理します。これにはイベント通知の設定が必要ですが、同期ごとにスコープ内の全オブジェクトをリストアップするという一般的なアプローチに比べ、はるかに堅牢で性能に優れた代替手段となります。全件リストアップは、オブジェクト数が数百万件に及ぶとスケールしにくく、コストもすぐに跳ね上がってしまいます。

仕組み #

暗黙の順序を持たないファイルを取り込むよう S3 ClickPipe を設定するには、対象のバケットに接続された Amazon Simple Queue System (SQS) キューを準備する必要があります。これにより、ClickPipes は指定されたパスに一致するキュー内の ObjectCreated イベントを監視できるようになります。

順不同モード: ファイルは任意の順序で S3 に配置され (A)、SQS 通知がトリガーされます (B-C)。ClickPipes は指定されたパスのファイルをポーリングして処理し、メタデータストアを使用して状態を追跡します (1-5)。データは exactly-once (正確に1回) の保証付きでターゲットテーブルに挿入されます (6)。

「障害が発生した場合はどうなるのか？」 上記のステップは複数のシステムにまたがっており、1つの巨大なトランザクションとして実行されるわけではないため、S3 からの読み込み、ファイルの処理済みマーク付け、ClickHouse へのデータ挿入など、あらゆるステップで障害が発生する可能性があります。障害が発生した場合、ClickPipes はバッチを自動的に再処理します。「重複はどうなるのか？」 ファイルが複数回再処理された場合でも、S3 ClickPipe は exactly-once セマンティクスを保証するため、ターゲットテーブルに重複が入り込むことはありません。

それでは、実際の動作を見ていきましょう。

Amazon SQS キューの作成 #

以下の手順は、データが入った S3 バケットがすでに存在し、AWS アカウント内で IAM ロールの管理や新しいリソースの作成を行う十分な権限があることを前提としています。

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

2. キューの信頼ポリシーを編集し、AWS アカウント (<aws-account-id>) 内のバケット (<bucket-arn>) から SQS キュー (<queue-arn>) へのメッセージ送信を許可します。

1{
2    "Version":"2012-10-17",                   
3    "Id": "example-ID",
4    "Statement": [
5        {
6            "Sid": "AllowS3ToSendMessage",
7            "Effect": "Allow",
8            "Principal": {
9                "Service": "s3.amazonaws.com"
10            },
11            "Action": [
12                "SQS:SendMessage"
13            ],
14            "Resource": "<sqs-queue-arn>",
15            "Condition": {
16                "ArnLike": {
17                    "aws:SourceArn": "<bucket-arn>"
18                },
19                "StringEquals": {
20                    "aws:SourceAccount": "<aws-account-id>"
21                }
22            }
23        }
24    ]
25}

3. ClickHouse Cloud コンソールで、Settings > Network security information に移動し、お使いのサービスの IAM ロール ARN をコピーします。

4. AWS コンソールに戻り、IAM > ロール > ロールの作成に移動します。カスタム信頼ポリシーを選択し、ClickHouse Cloud サービスの IAM ロール ARN を貼り付けます。

1{
2  "Version": "2012-10-17",
3  "Statement": [
4    {
5      "Sid": "AllowAssumeRole",
6      "Effect": "Allow",
7      "Principal": {
8        "AWS": "<ch-cloud-arn>"
9      },
10      "Action": "sts:AssumeRole"
11    }
12  ]
13}

5. S3 バケットからのオブジェクトのスキャンおよび取得と、SQS キュー内のメッセージを管理するために必要な権限を持つインラインポリシーを IAM ロールに作成します。

1{
2  "Version": "2012-10-17",
3  "Statement": [
4    {
5      "Sid": "S3BucketMetadataAccess",
6      "Effect": "Allow",
7      "Action": [
8        "s3:GetBucketLocation",
9        "s3:ListBucket"
10      ],
11      "Resource": "<bucket-arn>"
12    },
13    {
14      "Sid": "AllowGetListObjects",
15      "Effect": "Allow",
16      "Action": [
17        "s3:Get*",
18        "s3:List*"
19      ],
20      "Resource": "<bucket-arn>/*"
21    },
22    {
23      "Sid": "SQSNotificationsAccess",
24      "Effect": "Allow",
25      "Action": [
26        "sqs:DeleteMessage",
27        "sqs:ListQueues",
28        "sqs:ReceiveMessage",
29        "sqs:GetQueueAttributes"
30      ],
31    "Resource": "<sqs-queue-arn>"
32    }
33  ]
34}

IAM の設定項目は多かったですが、これで新しいファイルが届くたびにバケットから継続的にデータを取り込む ClickPipe を作成する準備が整いました。

ClickPipe の作成 #

1. ClickHouse Cloud コンソールで、Data Sources > Create ClickPipe に移動し、Amazon S3 を選択します。前のセクションで作成した ARN ロールを認証方法として IAM role を使い、S3 バケットに接続するための詳細情報を入力します。

2. Incoming data をクリックします。Continuous ingestion をオンに切り替えると、新しい Any order 取り込みオプションが表示されます。

3. Parse information をクリックします。ターゲットテーブルの Sorting key を定義し (非常に重要です)、マッピングされたスキーマに必要な調整を加えます。最後に、ClickPipes データベースユーザーのロールを設定します。

4. あとは待つだけです。ClickPipes がバケットの初期スキャンを実行し、新しい通知イベントが届き次第、ファイルの処理を開始します。🚀

クリック操作が多いと感じられるかもしれませんが、ClickPipes は ClickHouse Terraform プロバイダー で完全にサポートされているため、このセットアップをコードとして最初から最後まで構成できます。

今後の展望 #

S3 ClickPipes の新しい順不同モードにより、オブジェクトストレージから ClickHouse Cloud へのより柔軟な取り込みパターンが利用可能になりました。完全に管理された高速なエクスペリエンスはそのままに、さらなる柔軟性を手に入れられます。現時点では、この機能はデータソースとしての Amazon S3 のみで利用できます。すべての Object Storage ClickPipes 間で機能の同等性を確保するため、近い将来 Google Cloud Storage (GCS) および Azure Blob Storage (ABS) へのサポート拡張を予定しています。

ClickPipes のセットアップ中にフィードバックがある場合や問題が発生した場合は、私たちのチームまでお問い合わせください。ステップごとの詳細な手順、よくある質問、注意点については、S3 ClickPipes のドキュメント をご覧ください。