メインコンテンツまでスキップ
バージョン: 26.x

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 構成

フィールド説明例
Task Nameこのインポートジョブの一意の名前confluent_orders
Consumer GroupKafkaコンシューマーグループIDvelodb-consumer
Brokersブートストラップサーバー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 構成

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

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

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

ステップ 5: 宛先Tableを設定​

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

Data Preview

Data Previewセクションはトピックのサンプルレコードを表示します。

Table設定を構成​

Destination Table

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

カラム設定​

カラムマッピングを確認して調整します：

Column Settings

設定説明
Source FieldKafkaメッセージのJSONフィールド
Column NameVeloDBTableのカラム名
Data タイプVeloDBデータ型（自動検出）

詳細設定​

設定説明
Table Models追加専用の場合はDUPLICATE、アップサートの場合はUNIQUE
Sorting Keyデータ順序付け用のカラム
バケット Keyデータ分散用のカラム
バケット 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 許可 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 Jobs の管理​

アクション方法
一時停止ジョブをクリックし、Pause をクリック
再開ジョブをクリックし、Resume をクリック
削除ジョブをクリックし、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を確認してください

参考資料​