StreamkapをClickHouseに接続する

Streamkap は、ストリーミング CDC (変更データキャプチャ) とストリーム処理に特化したリアルタイムデータインテグレーションプラットフォームです。Apache Kafka、Apache Flink、Debezium を活用した高スループットかつスケーラブルなスタック上に構築されており、SaaS または BYOC (Bring Your Own Cloud) 形態の完全マネージド型サービスとして提供されています。

Streamkap を使用すると、PostgreSQL、MySQL、SQL Server、MongoDB、そのほか多数のソースデータベースで発生するあらゆる insert、update、delete を、ミリ秒単位のレイテンシで直接 ClickHouse にストリーミングできます。

そのため、リアルタイム分析ダッシュボード、オペレーショナルアナリティクス、機械学習モデルへのライブデータ供給に最適です。

• リアルタイムストリーミング CDC: Streamkap はデータベースのログから変更を直接取り込み、ClickHouse 内のデータがソースのリアルタイムなレプリカであることを保証します。 シンプルなストリーム処理: ClickHouse に取り込む前に、データをリアルタイムで変換、エンリッチ、ルーティング、フォーマットし、embeddings を作成できます。複雑さを伴わない Flink を基盤としています

• 完全マネージド型でスケーラブル: 本番環境に対応した、メンテナンス不要のパイプラインを提供し、Kafka、Flink、Debezium、またはスキーマレジストリのインフラストラクチャを自前で管理する必要をなくします。このプラットフォームは高スループット向けに設計されており、数十億件のイベントを処理できるようリニアにスケールします。

• 自動スキーマ進化: Streamkap はソースデータベースのスキーマ変更を自動的に検出し、それらを ClickHouse に反映します。手動で介入しなくても、新しいカラムの追加やカラム型の変更に対応できます。

• ClickHouse 向けに最適化: このインテグレーションは、ClickHouse の機能を効率的に活用できるように構築されています。既定では、ReplacingMergeTree エンジンを使用して、ソースシステムからの更新や削除をシームレスに処理します。

• 耐障害性の高い配信: このプラットフォームは at-least-once 配信保証を提供し、ソースと ClickHouse 間のデータ整合性を確保します。upsert 操作では、主キーに基づいて重複排除を行います。

このガイドでは、Streamkap パイプラインを設定して ClickHouse にデータを読み込む方法の概要を説明します。

• Streamkap アカウント。
• ClickHouse クラスターの接続情報: ホスト名、Port、Username、Password。
• CDC (変更データキャプチャ) を有効にするよう設定されたソースデータベース (例: PostgreSQL、SQL Server) 。詳細なセットアップガイドは Streamkap のドキュメントで確認できます。

SELECT * FROM your_table_name LIMIT 10;

Streamkapのインテグレーションは、ClickHouse内のCDC (変更データキャプチャ) データを効率的に管理できるように設計されています。

デフォルトでは、Streamkap は upsert インジェストモードを使用します。ClickHouse にテーブルを作成する際には、ReplacingMergeTree エンジンが使われます。このエンジンは、CDC イベントの処理に適しています。

• ソーステーブルの主キーは、ReplacingMergeTree のテーブル定義で ORDER BY キーとして使用されます。

• ソースでの 更新 は、ClickHouse では新しい行として書き込まれます。バックグラウンドのマージ処理で、ReplacingMergeTree はこれらの行をまとめ、ソートキーに基づいて最新バージョンだけを残します。

• 削除 は、ReplacingMergeTree の is_deleted パラメータに渡されるメタデータフラグによって処理されます。ソースで削除された行はすぐには削除されず、削除済みとしてマークされます。

  • 必要に応じて、削除済みレコードを分析目的で ClickHouse に保持できます

Streamkap は、データの状態を管理するために各テーブルに複数のメタデータカラムを追加します。

ReplacingMergeTree は更新と削除をバックグラウンドで処理するため、単純な SELECT * クエリでは、マージが完了する前の古い行や削除済みの行が表示されることがあります。データの最新の状態を取得するには、削除されたレコードを除外し、各行の最新バージョンのみを選択する必要があります。

これには FINAL 修飾子を使用できます。便利ではありますが、クエリのパフォーマンスに影響する可能性があります。

-- FINALを使用して正しい現在の状態を取得する
SELECT * FROM your_table_name FINAL WHERE __DELETED = 'false';
SELECT * FROM your_table_name FINAL LIMIT 10;
SELECT * FROM your_table_name FINAL WHERE <filter by keys in ORDER BY clause>;
SELECT count(*) FROM your_table_name FINAL;

大規模なテーブルでパフォーマンスを向上させるには、特にすべてのカラムを読み取る必要がない場合や単発の分析クエリでは、argMax関数を使って各主キーの最新レコードを手動で選択できます。

SELECT key,
       argMax(col1, version) AS col1,
       argMax(col2, version) AS col2
FROM t
WHERE <フ

本番環境のユースケースや、エンドユーザーから同時に繰り返し実行されるクエリに対しては、後続のアクセスパターンにより適した形でデータを構成するために、materialized viewを使用できます。

• Streamkap 公式サイト
• ClickHouse 向け Streamkap ドキュメント
• ブログ: 変更データキャプチャを使用した ClickHouse へのストリーミング
• ClickHouse ドキュメント: ReplacingMergeTree