Amazon Kinesis と ClickHouse Cloud の統合

Kinesis ClickPipes は、ClickPipes UI を使用して手動でデプロイおよび管理できるほか、OpenAPI や Terraform を使用してプログラムによってデプロイおよび管理することもできます。 あらかじめ ClickPipes の概要を確認し、IAM 認証情報または IAM ロールを設定しておいてください。ClickHouse Cloud で使用するロールの設定方法については、Kinesis のロールベースアクセスガイドを参照してください。

​

最初の ClickPipe を作成する

1. ClickHouse Cloud サービスの SQL Console にアクセスします。

2. 左側のメニューで Data Sources ボタンを選択し、“Set up a ClickPipe” をクリックします。

3. データソースを選択します。

4. ClickPipe の名前、説明 (任意) 、IAM ロールまたは認証情報、そのほかの接続情報を入力します。

5. Kinesis ストリーム、Data format、開始オフセットを選択します。Protobuf の場合は、.proto ファイルまたはシリアライズされた FileDescriptorSet (最大 768 KiB) をアップロードします。スキーマレジストリは不要です。 UI には、選択した stream のサンプルデータが表示されます。Use Kinesis Enhanced Fan-Out はデフォルトで有効になっています。詳細は Amazon Kinesis のドキュメントを参照してください。

6. 次のステップでは、データの取り込み先を新しい ClickHouse テーブルにするか、既存のテーブルを再利用するかを選択できます。画面の指示に従って、テーブル名、スキーマ、設定を変更してください。変更内容は、上部のサンプルテーブルでリアルタイムにプレビューできます。

用意されているコントロールを使って、高度な設定をカスタマイズすることもできます。

7. また、既存の ClickHouse テーブルにデータを取り込むこともできます。その場合、UI ではソースのフィールドを、選択した宛先テーブル内の ClickHouse フィールドにマッピングできます。

8. 最後に、内部 ClickPipes ユーザーの権限を設定できます。

Permissions: ClickPipes は、宛先テーブルにデータを書き込むための専用ユーザーを作成します。カスタムロールまたは定義済みロールのいずれかを使って、この内部ユーザーに割り当てるロールを選択できます。

• Full access: クラスターへのフルアクセス権を持ちます。宛先テーブルで materialized view や Dictionary を使用する場合に便利です。
  • Only destination table: 宛先テーブルに対する INSERT 権限のみを持ちます。

9. “Complete Setup” をクリックすると、システムに ClickPipe が登録され、サマリーテーブルに表示されるようになります。

サマリーテーブルには、ソースまたは ClickHouse 内の宛先テーブルからサンプルデータを表示するためのコントロールがあります。 また、ClickPipe を削除したり、インジェストジョブの概要を表示したりするためのコントロールもあります。

10. おめでとうございます！ 最初の ClickPipe の設定は正常に完了しました。これが streaming ClickPipe の場合は継続的に動作し、リモートデータソースからデータをリアルタイムで取り込み続けます。それ以外の場合は、バッチを取り込んで完了します。

​

対応データフォーマット

対応フォーマットは以下のとおりです。

• JSON
• Protobuf

ClickPipes for Kinesis は、圧縮されたレコードを自動的に検出して解凍します。Kafka ではクライアントライブラリが透過的に解凍を処理しますが、Kinesis は生のバイト列をそのまま配信するため、これは ClickPipes が設定不要で処理します。 サポートされている圧縮コーデックは次のとおりです。

• gzip
• zstd
• lz4
• snappy (フレーム形式)

圧縮は、各レコード内のマジックバイトによって自動的に検出されます。既知の圧縮シグネチャが見つからない場合、そのレコードは非圧縮として扱われます。検出された圧縮タイプはスキーマ推論時にも表示されるため、UI のサンプルデータプレビューでも解凍後のデータが正しく表示されます。 現在、ClickPipes では以下の ClickHouse データ型がサポートされています。

• 基本的な数値型 - [U]Int8/16/32/64、Float32/64、BFloat16
• 大きな整数型 - [U]Int128/256
• Decimal 型
• Boolean
• String
• FixedString
• Date、Date32
• DateTime、DateTime64 (UTC タイムゾーンのみ)
• Enum8/Enum16
• UUID
• IPv4
• IPv6
• すべての ClickHouse LowCardinality 型
• 上記のいずれかの型をキーと値に使用する Map (Nullable を含む)
• 上記のいずれかの型を要素に使用する Tuple および Array (Nullable を含む、深さは 1 レベルのみ)
• SimpleAggregateFunction 型 (AggregatingMergeTree または SummingMergeTree の宛先向け)

ソースデータストリーム内の任意の JSON フィールドに対して、Variant 型 (Variant(String, Int64, DateTime) など) を手動で指定できます。 ClickPipes では使用する適切な Variant 型を判定する仕組み上、Variant の定義で使用できる整数型または datetime 型は 1 つだけです。たとえば、Variant(Int64, UInt32) はサポートされていません。 常に JSON オブジェクトである JSON フィールドは、JSON の宛先カラムに割り当てることができます。固定パスやスキップするパスも含め、宛先カラムを目的の JSON 型に手動で変更する必要があります。 Kinesis ストリームでは、以下の仮想カラムをサポートしています。新しい宛先テーブルを作成する際は、Add Column ボタンを使って仮想カラムを追加できます。 _raw_message フィールドは、Kinesis の JSON レコード全体だけが必要な場合に使用できます (たとえば、ClickHouse の JsonExtract* 関数を使用して、下流の materialized view ビューにデータを投入する場合) 。このようなパイプでは、「仮想カラムではない」カラムをすべて削除することで、ClickPipes のパフォーマンスが向上する場合があります。

• DEFAULT はサポートされていません。
• 最小の (XS) レプリカサイズで実行する場合、個々のメッセージの上限はデフォルトで 16MB (非圧縮) 、より大きいレプリカでは 32MB (非圧縮) です。この上限を超えるメッセージは error となり、拒否されます。より大きいメッセージが必要な場合は、サポートにお問い合わせください。

ClickPipes はデータをバッチ単位で ClickHouse に挿入します。これは、データベース内に過剰な数のパーツが作成されるのを防ぎ、クラスターのパフォーマンス低下を避けるためです。 バッチは、次のいずれかの条件を満たした時点で挿入されます。

• バッチサイズが上限に達した場合 (100,000 行、またはレプリカメモリ 1GB あたり 32MB)
• バッチの保持時間が上限に達した場合 (5 秒)

レイテンシ (Kinesis のメッセージがストリームに送信されてから、そのメッセージを ClickHouse で利用できるようになるまでの時間) は、さまざまな要因 (例: Kinesis のレイテンシ、ネットワークレイテンシ、メッセージのサイズ/フォーマット) に左右されます。上のセクションで説明したバッチ処理も、レイテンシに影響します。実際にどの程度のレイテンシになるかを把握するため、必ずご利用のユースケースに即してテストすることをお勧めします。 低レイテンシに関する特定の要件がある場合は、お問い合わせください。 アクティブな分片が多数ある場合、それらを処理できるだけのスケールが ClickPipe に設定されていないと、ClickPipe のスループットが低下する可能性があります。 ClickPipes for Kinesis は、水平方向と垂直方向の両方にスケールできるよう設計されています。デフォルトでは、1 つのコンシューマーを含むコンシューマグループが作成されます。これは ClickPipe の作成時、または作成後の任意のタイミングで Settings -> Advanced Settings -> Scaling から設定できます。 ClickPipes は、アベイラビリティゾーンに分散されたアーキテクチャによって高可用性を実現します。 そのため、少なくとも 2 つのコンシューマーまでスケールする必要があります。 実行中のコンシューマー数にかかわらず、耐障害性は設計上確保されています。 コンシューマーまたはその基盤となるインフラストラクチャで障害が発生した場合、 ClickPipe は自動的にコンシューマーを再起動し、メッセージの処理を継続します。

​

認証

Amazon Kinesis ストリームにアクセスするには、IAM認証情報 または IAM ロール を使用できます。IAM ロール の設定方法の詳細については、ClickHouse Cloud で利用できるロールの設定手順を説明したこちらのガイドを参照してください