Kafka

データを作成する

1. Kafka ツールウィンドウを開きます: 表示 | ツールウィンドウ | Kafka。

2. (新規接続) をクリックします。

3. Kafka 接続を選択し、 プロデューサー},{ をクリックします。

   これにより、新しいエディタータブでプロデューサーが開きます。

4. トピック リストで、メッセージを書き込むトピックを選択します。

5. キー および 値 で、メッセージキーと値を選択します。

   スキーマレジストリに接続している場合は、 スキーマレジストリ を選択して、送信されたデータを選択したスキーマと照合してチェックできます。

   ランダムな値を生成できます:

   • をクリックすると、選択したタイプに基づいてランダムな値が生成されます。 これには、選択したスキーマレジストリに基づいて JSON オブジェクト全体の生成が含まれます。

   • ランダムな値を生成する柔軟性を高めるには、 ${random...} 変数を使用します。 JSON、Avro、Protobuf ファイルを編集するときに、値フィールドに random と入力し始めると、ランダムな値の自動補完オプションが表示されます。 例: "${random.integer(1,10)}" を使用して、1 から 10 までのランダムな整数を生成できます。

     

6. ヘッダー で、カスタムヘッダーを指定します。 JSON または CSV 形式で保存されている場合は、このセクションに貼り付けることができます。

7. フロー では、レコードフローを制御できます。

   • 複数のレコードを同時に送信する場合は、 一度に記録 に数値を入力します。

   • レコードデータをランダムに生成する場合は、 ランダムなキーを生成する および ランダムな値を生成する を選択します。

   • レコード送信間の 間隔 をミリ秒単位で設定します。

   • 指定されたレコード数に達した場合、または指定された時間が経過した場合に、プロデューサーがメッセージの送信を停止するようにするには、 停止条件 を指定します。

8. オプション で、追加のオプションを指定します。

   • パーティション ：レコードを送信するトピックパーティションを指定します。 指定されていない場合は、デフォルトのロジックが使用されます：プロデューサーが鍵のハッシュ値をパーティション数で割った値を使います。

   • 圧縮: プロデューサーによって生成されたデータの圧縮タイプを選択します: なし、 Gzip、 Snappy、 Lz4 、または Zstd。

   • 冪等性: 各メッセージのコピーが 1 つだけストリームに書き込まれるようにする場合に選択します。

   • ACK: リーダーがローカルログにレコードを書き込み、すべてのフォロワーからの完全な確認を待たずに応答するようにする場合は、 リーダー を選択します。 リーダーが同期レプリカの完全なセットがレコードを確認するのを待機するには、 すべて を選択します。 サーバーからの確認応答を待たないように、プロデューサー用の なし を保持します。

9. 生産 をクリックします。

データ タブ内の任意のレコードをクリックすると、その詳細が表示されます。 をクリックして統計を有効にすることもできます。

データを消費する

1. Kafka ツールウィンドウを開きます: 表示 | ツールウィンドウ | Kafka。

2. (新規接続) をクリックします。

3. Kafka 接続を選択し、 消費者 をクリックします。

   これにより、新しいエディタータブでコンシューマーが開きます。

4. トピック リストで、サブスクライブするトピックを選択します。

5. キー および 値 で、使用するレコードのキーと値のデータ型を選択します。

6. 範囲とフィルター を使用して、使用するデータを絞り込みます。

   • 開始 リストで、データを消費する期間またはオフセットを選択します。 トピックからすべてのレコードを取得するには、 先頭から を選択します。

   • 制限 リストで、データの受信をいつ停止するかを選択します (たとえば、トピック内のレコード数が特定の数に達したとき)。

   • フィルター を使用して、キー、値、またはヘッダーの部分文字列によってレコードをフィルターします。

7. その他 で、追加のパラメーターを設定します。

   • パーティション ボックスにパーティション ID またはコンマ区切りの ID リストを入力して、特定のパーティションからのレコードのみを取得します。

   • 新しいコンシューマーを追加する場合は、 消費者団体 リストでコンシューマーグループを選択します。

8. 消費を開始する をクリックします。

データ タブ内の任意のレコードをクリックすると、その詳細が表示されます。 をクリックして統計を有効にすることもできます。

データのエクスポート

生成または消費されたデータを CSV、TSV、JSON 形式でダウンロードできます。

1. データの 生成または 消費を開始します。

2. データ テーブルの右上部分で、 をクリックし、 CSV、 TSV、 JSON を選択します。

3. 出力ファイルの場所を選択し、 保存する をクリックします。

プロデューサーまたはコンシューマーのプリセットを保存する

同じキー、値、ヘッダー、またはその他のパラメーターを使用してデータを頻繁に生成または使用する場合は、プリセットとして保存できます。 その後、プリセットを再利用して、プロデューサーまたはコンシューマーをすばやく作成できます。

1. Kafka ツールウィンドウで、 プロデューサー},{ または 消費者 をクリックします。

2. 必要なパラメーターを指定し、プロデューサーまたはコンシューマー作成フォームの上部で (プリセットの保存) をクリックします。

パラメーターはプリセットとして保存され、 プリセット タブで利用できます。 プリセットをクリックして適用します。

スキーマレジストリに接続する

1. クラウドプロバイダー、 カスタムサーバー 、または プロパティを使用して、Kafka ブローカーへの接続を作成します。

2. Confluent を使用する場合は、ブローカーレジストリプロパティとスキーマレジストリプロパティの両方を 構成 フィールドに貼り付けることができます。

   それ以外の場合は、「スキーマレジストリ 」セクションを展開し、プロバイダー (Confluent または Glue) を選択します。

   • URL: スキーマレジストリの URL を入力します。

   • 構成ソース: 接続パラメーターを提供する方法を選択します。

     • カスタム: 認証方法を選択し、資格情報を提供します。

       Kafka ブローカーの SSL 設定とは異なる SSL 設定を使用する場合は、 ブローカー SSL 設定を使用する チェックボックスをオフにして、トラストストアのパスを指定します。

     • プロパティ: 提供された構成プロパティを貼り付けます。 または、コード補完と IntelliJ IDEA が提供する簡単なドキュメントを使用して、プロパティを手動で入力することもできます。

   • 地域: スキーマレジストリリージョンを選択します。

   • AWS 認証: 認証方法を選択します:

     • デフォルトの資格情報プロバイダーチェーン: デフォルトのプロバイダーチェーンの資格情報を使用します。 チェーンの詳細については、「デフォルトの資格情報プロバイダーチェーンの使用(英語) 」を参照してください。

     • 資格情報ファイルからのプロファイル: credentials ファイルからプロファイルを選択します。

     • 明示的なアクセスキーと秘密鍵: 資格情報を手動で入力します。

   • レジストリ名: 接続するスキーマレジストリの名前を入力するか、 をクリックしてリストから選択します。

3. 設定を入力したら、 接続のテスト をクリックして、すべての構成パラメーターが正しいことを確認します。 次に OK をクリックします。

スキーマの作成

1. Kafka ツールウィンドウを開きます: 表示 | ツールウィンドウ | Kafka。

2. (新規接続) をクリックします。

3. スキーマレジストリ を選択し、 をクリックします (または Alt+Insert を押します)。

4. フォーマット リストで、スキーマ形式 (Avro、Protobuf、JSON) を選択します。

5. 戦略 リストで、 命名戦略(英語)を選択し、選択した戦略に応じて、名前接尾辞を設定するか、トピックを選択します。 または、 カスタム名 を選択し、任意の名前を入力します。

スキーマをツリーおよび生のビューでプレビューできます。

スキーマのバージョンを比較する

1. スキーマレジストリに接続する場合は、 スキーマレジストリ のスキーマを選択します。

2. 生のビュー に切り替えて、 比較 をクリックします。 このボタンは、スキーマに複数のバージョンがある場合に使用できます。

スキーマのバージョンを削除する

スキーマに複数のバージョンがある場合、特定のバージョンを削除できます。 スキーマレジストリは、 ソフト削除（バージョンの削除後もスキーマのメタデータと ID がレジストリから削除されない）とハード削除（スキーマ ID を含むすべてのメタデータが削除される）の 2 種類の削除(英語)をサポートしています。 選択できる方法は、Confluent スキーマレジストリと AWS Glue スキーマレジストリのどちらを使用しているかによって異なります。

• Confluent Schema Registry では、デフォルトで論理的な削除が使用されます。 完全な削除 チェックボックスを選択すると、完全な削除の使用を選択できます。

• AWS Glue Schema Registry は常に完全削除を使用します。

1. スキーマレジストリ でスキーマを選択します。

2. その右側にある をクリックし、 バージョンの削除 を選択します。