Confluent Cloud への接続
このガイドでは、ビジュアルインターフェースを使用してVeloDB CloudをConfluent Cloud Kafkaに接続する手順を説明します。
前提条件: 先に進む前に、Confluent Cloud Setup Guideを完了してクラスター、APIキー、サンプルデータトピックを作成してください。
ステップ1: Importへ移動
VeloDBウェアハウスで、左サイドバーのDataセクションに移動し、Importをクリックします。
Createをクリックして新しいインポートジョブを開始します。
ステップ2: Confluent Cloudを選択
Event Streamsの下で、Confluent Cloudをクリックしてストリーミングインポートウィザードを開始します。
ステップ3: 接続の設定
Confluent Cloud Setupからの接続詳細を入力します：
| フィールド | 説明 | 例 |
|---|---|---|
| Task Name | このインポートジョブの一意な名前 | confluent_orders |
| Consumer Group | KafkaコンシューマーグループID | velodb-consumer |
| Brokers | BootstrapサーバーURL | pkc-xxxxx.us-east-1.aws.confluent.cloud:9092 |
| API Key | Confluent Cloud APIキー | あなたのAPIキー |
| API Secret | Confluent Cloud APIシークレット | あなたのAPIシークレット |
| SASL Mechanism | 認証方法 | PLAIN |
すべてのフィールドを入力した後、Nextをクリックします。
ステップ4: ソースデータの選択
Kafkaトピックとデータフォーマットを設定します：
| フィールド | 説明 |
|---|---|
| Topic | ドロップダウンからKafkaトピックを選択 |
| Offset Selection | すべてのデータをロードする場合はFrom beginning、新しいデータのみの場合はFrom latest |
| Data Format | JSONを選択 |
VeloDBはJSONとCSVフォーマットのみをサポートしています。ConfluentトピックでAVROを使用している場合、Confluent CloudでJSONフォーマットの新しいトピックを作成する必要があります。
Nextをクリックして続行します。
ステップ5: 宛先テーブルの設定
VeloDBはKafkaメッセージからスキーマを自動検出します：
Data Previewセクションはトピックからのサンプルレコードを表示します。
テーブル設定の構成
| フィールド | 説明 |
|---|---|
| Load Data to | 新しいテーブルを作成する場合はNew Tableを選択 |
| Database | データベースを選択または作成 |
| Table | テーブル名を入力 |
カラム設定
カラムマッピングを確認し調整します：
| 設定 | 説明 |
|---|---|
| Source Field | KafkaメッセージのJSONフィールド |
| Column Name | VeloDBテーブルカラム名 |
| Data Type | VeloDBデータタイプ（自動検出） |
詳細設定
| 設定 | 説明 |
|---|---|
| Table Models | 追記専用の場合はDUPLICATE、アップサートの場合はUNIQUE |
| Sorting Key | データ順序付け用のカラム |
| Bucket Key | データ分散用のカラム |
| Bucket Number | AUTO推奨 |
Nextをクリックして続行します。
ステップ6: 設定の構成
インポートジョブ設定を調整します：
| 設定 | デフォルト | 説明 |
|---|---|---|
| Concurrency | 256 | 並列コンシューマー数 |
| Max Batch Interval (s) | 60 | コミット前の最大待機時間 |
| Max Batch Rows | 20000000 | バッチあたりの最大行数 |
| Max Batch Size (MB) | 1024 | 最大バッチサイズ |
デフォルトはほとんどのユースケースで適用できます。Nextをクリックして続行します。
ステップ7: 検証
VeloDBが設定を検証します：
チェックリストで以下を検証します：
- Kafka Validation - Confluent Cloudへの接続
- Warehouse Connectivity Test - VeloDBがKafkaに到達できるか
- User Permission Check - インポートを作成する権限があるか
すべてのチェックがSucceededを表示したら、Startをクリックしてデータストリーミングを開始します。
データインポートの検証
インポート開始後、データが流れているか確認します：
インポートステータスの確認
サイドバーのImportに移動してジョブステータスを確認します：
- RUNNING - ジョブがアクティブにデータを消費中
- PAUSED - ジョブが一時停止中（エラーをチェック）
データのクエリ
SQL Editorに移動して実行します：
-- Check row count
SELECT COUNT(*) FROM your_database.your_table;
-- View sample data
SELECT * FROM your_database.your_table LIMIT 10;
Import Jobの管理
| アクション | 方法 |
|---|---|
| 一時停止 | jobをクリックし、Pauseをクリック |
| 再開 | jobをクリックし、Resumeをクリック |
| 削除 | jobをクリックし、Deleteをクリック |
またはSQLを使用：
-- Pause job
PAUSE ROUTINE LOAD FOR database.job_name;
-- Resume job
RESUME ROUTINE LOAD FOR database.job_name;
-- Stop job
STOP ROUTINE LOAD FOR database.job_name;
-- View job status
SHOW ROUTINE LOAD FOR database.job_name;
トラブルシューティング
| 問題 | 解決方法 |
|---|---|
| "Incorrect credentials" | Confluent CloudからAPI KeyとSecretを確認してください |
| "Broker transport failure" | SASL MechanismがPLAINに設定されていることを確認してください |
| "Topic not found" | トピック名が正確に一致していることを確認してください（大文字小文字を区別） |
| JSON parse error | ConfluentトピックがAVROではなくJSON形式を使用していることを確認してください |
| Job paused with errors | エラーの詳細についてはSHOW ROUTINE LOADを確認してください |
参考資料
- Confluent Cloud Setup Guide
- Kafka Integration Guide - SQLベースのセットアップとその他のKafkaプロバイダーについて
- CREATE ROUTINE LOAD