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. Flow では、レコードフローを制御できます。

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

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

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

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

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 設定を使用するチェックボックスをオフにして、トラストストアのパスを指定します。

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

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

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

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

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

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

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

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. その右側にある をクリックし、バージョンの削除を選択します。