メインコンテンツまでスキップ

Confluent Cloudに接続

このガイドでは、ビジュアルインターフェースを使用してVeloDB CloudをConfluent Cloud Kafkaに接続する手順を説明します。

警告

前提条件: 続行する前に、Confluent Cloud Setup Guideを完了して、クラスター、APIキー、サンプルデータトピックを作成してください。

ステップ1: Importに移動​

VeloDBウェアハウスで、左サイドバーのDataセクションに移動し、Importをクリックします。

Createをクリックして新しいインポートジョブを開始します。

New Import Page

ステップ2: Confluent Cloudを選択​

Event Streamsの下で、Confluent Cloudをクリックしてストリーミングインポートウィザードを開始します。

ステップ3: 接続を設定​

Confluent Cloud Setupからの接続詳細を入力します：

Connection Configuration

フィールド説明例
Task Nameこのインポートジョブの一意な名前confluent_orders
Consumer GroupKafkaコンシューマーグループ IDvelodb-consumer
BrokersBootstrapサーバーURLpkc-xxxxx.us-east-1.aws.confluent.cloud:9092
API KeyConfluent Cloud APIキーYour API Key
API SecretConfluent Cloud APIシークレットYour API Secret
SASL Mechanism認証方式PLAIN

すべてのフィールドを入力後、Nextをクリックします。

ステップ4: ソースデータを選択​

Kafkaトピックとデータ形式を設定します：

Source Data Configuration

フィールド説明
TopicドロップダウンからKafkaトピックを選択
Offset Selectionすべてのデータをロードする場合はFrom beginning、新しいデータのみの場合はFrom latest
Data FormatJSONを選択
Data Format

VeloDBはJSONとCSV形式のみをサポートします。ConfluentトピックでAVROを使用している場合は、Confluent CloudでJSON形式の新しいトピックを作成する必要があります。

Nextをクリックして続行します。

ステップ5: 宛先テーブルを設定​

VeloDBはKafkaメッセージからスキーマを自動検出します：

Data Preview

Data Previewセクションでは、トピックのサンプルレコードが表示されます。

テーブル設定を構成​

Destination Table

フィールド説明
Load Data to新しいテーブルを作成するにはNew Tableを選択
Databaseデータベースを選択または作成
Tableテーブル名を入力

列設定​

列マッピングを確認し調整します：

Column Settings

設定説明
Source FieldKafkaメッセージからの JSONフィールド
Column NameVeloDBテーブルの列名
Data TypeVeloDBデータタイプ（自動検出）

詳細設定​

設定説明
Table Models追加専用の場合はDUPLICATE、アップサートの場合はUNIQUE
Sorting Keyデータ順序付けのための列
Bucket Keyデータ分散のための列
Bucket NumberAUTOを推奨

Nextをクリックして続行します。

ステップ6: 設定を構成​

インポートジョブの設定を調整します：

Settings

設定デフォルト説明
Concurrency256並列コンシューマー数
Max Batch Interval (s)60コミット前の最大待機時間
Max Batch Rows20000000バッチあたりの最大行数
Max Batch Size (MB)1024最大バッチサイズ

デフォルト値はほとんどのユースケースで適切です。Nextをクリックして続行します。

ステップ7: 検証​

VeloDBは設定を検証します：

Verification

チェックリストは以下を検証します：

  • 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 errorConfluentトピックがAVROではなくJSON形式を使用していることを確認してください
Job paused with errorsエラーの詳細についてはSHOW ROUTINE LOADを確認してください

参考資料​