Kafka

データを作成する

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

2. Kafka 接続を選択し、 ( プロデューサーの作成 ) をクリックします。

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

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

4. キーおよび値で、メッセージのキーと値を選択します。 を使用して、選択したタイプに基づいてランダムな値を生成します。

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

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

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

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

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

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

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

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

   • パーティション : レコードを送信する必要があるトピックパーティションを指定します。指定しない場合は、デフォルトのロジックが使用されます。プロデューサーは、パーティション数を法としたキーのハッシュを取得します。

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

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

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

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

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

データを消費する

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

2. Kafka 接続を選択し、 ( コンシューマーの作成 ) をクリックします。

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

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

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

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

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

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

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

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

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

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

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

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

1. Kafka ツールウィンドウで、 ( プロデューサーの作成 ) または ( コンシューマーの作成 ) をクリックします。

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

スキーマの作成

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

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

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

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

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

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

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

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

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

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

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

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

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

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