1. Home
2. Platform
3. スキーマ管理

トピックのスキーマを管理する

Control Center 内の Schema Registry 機能を使用して Confluent Platform トピックのスキーマを管理します。

以下の操作を行うことができます。

• スキーマの作成、編集、表示
• スキーマのバージョンの比較
• スキーマのダウンロード

Schema Registry では、スキーマに対する検証および互換性チェックを実行します。

ちなみに

• Schema Registry を使用するための ステップバイステップのチュートリアル を確認します。
• Schema Registry の詳細については、Schema Registry ドキュメント を参照してください。

Control Center 内の Schema Registry 機能は、デフォルトで 有効になります。この機能を無効にすると、スキーマの表示と編集の両方が無効になります。

注釈

Confluent Platform バージョン 5.2 以降では、スキーマを作成および編集できます。5.2 より前のバージョンでは、バージョン履歴の比較やスキーマのダウンロードなど、Control Center 内にあるスキーマ情報への表示アクセスのみが提供されています。

Control Center でのトピックのスキーマの作成

キースキーマと値スキーマを作成します。通常は、値スキーマがキースキーマよりも高い頻度で作成されます。

ベストプラクティス:

• 後方互換性を容易にするために、フィールドにはデフォルト値を指定します（スキーマが関係する場合）。
• スキーマを人間が読めるようにするために、少なくともより不明瞭なフィールドを文書化します。

トピックの値スキーマを作成するには:

1. クラスターを選択します。

2. メニューの Topics をクリックします。All topics ページ が表示されます。

3. トピックを選択します。

   

   トピックの概要ページが表示されます。

   

4. Schema タブをクリックします。

   メッセージの値スキーマを設定するよう求められます。

   

5. Set a schema をクリックします。スキーマエディターが開き、必要に応じて開始点として使用できる Avro スキーマ の基本構造が事前に取り込まれて表示されます。

   

6. スキーマフォーマットのタイプを選択します。

   • Avro
   • JSON
   • Protobuf
   

   以降の手順で説明されているコード例を試すには Avro を選択します。

   ちなみに

   各スキーマタイプの詳細については、「サポートされているフォーマット」を参照してください。

7. スキーマをエディターで入力します。

   • name: デフォルトを受け入れない場合は、スキーマの名前を入力します。これは、サブジェクト名の戦略によって決定されます。デフォルトは、schema_type_topic_name です。これは必須です。

   • type: record、enum、union、array、map、または fixed のいずれか。（type に record を選択する場合はスキーマのトップレベルで指定され、データ型が異なる複数のフィールドを含めることができます。）これは必須です。

   • namespace: スキーマの命名の競合を防止するための完全修飾名。スキーマの name を修飾する文字列です。これは省略可能ですが、推奨されています。

   • fields: 1 つのレコードの 1 つ以上のフィールドをリストした JSON 配列。これは必須です。

     各フィールドには、以下の属性を指定できます。

     • name: フィールドの名前。これは必須です。
     • type: フィールドのデータ型。これは必須です。
     • doc: フィールドのメタデータ。これは省略可能ですが、推奨されています。
     • default: フィールドのデフォルト値。これは省略可能ですが、推奨されています。
     • order: フィールドのソート順序。有効な値は、ascending、descending、または ignore のいずれかです。デフォルト: Ascending。これは省略可能です。
     • aliases: フィールドの代替名。これは省略可能です。

   次のスキーマ例をコピーアンドペーストします。

   {
    "type": "record",
    "namespace": "my.examples",
    "name": "Payment",
    "fields": [
      {
       "name": "id",
       "type": "string"
      },
      {
       "name": "amount",
       "type": "double"
      }
              ]
   }
   

   Copy
   

8. （オプション）スキーマの作成前にその構文と構造を検証するには、Validate をクリックします。

9. Create をクリックします。

   • 入力したスキーマが有効な場合は、Schema updated というメッセージが短い間バナーエリアに表示されます。
   • 入力したスキーマが有効でない場合は、Input schema is an invalid Avro schema というエラーがバナーエリアに表示されます。

   この手順がトピックのキースキーマにも当てはまる場合は、必要に応じて手順を繰り返します。

トピックのキースキーマを作成するには:

1. Key オプションをクリックします。メッセージのキースキーマを設定するよう求められます。

   

2. Set a schema をクリックします。

3. スキーマをエディターに入力し、Save をクリックします。

   次のスキーマ例をコピーアンドペーストし、保存します。

   {
     "namespace": "io.confluent.examples.clients.basicavro",
     "name": "key_my_transactions",
     "type": "string"
   }
   

   Copy
   

ちなみに

Kafka のメッセージは、キー/値のペアです。メッセージのキーとメッセージの値は、別々にシリアライズすることができます。例えば、値では Avro record を使用し、キーではプリミティブ（string や integer など）を使用することもできます。一般的には、メッセージのキーを使用する場合はプリミティブになりますが、複合データ型になることもあります（例えば、record または array)。どのようにキーを設定するかは、お客様と、お客様の実装の要件によって異なります。キースキーマと値スキーマの詳細な例については、Schema Registry ドキュメントの フォーマット、シリアライザー、逆シリアライザー にある説明を参照してください。

Control Center でのスキーマの表示

特定のトピックのスキーマの詳細を表示します。

1. ナビゲーションバーでクラスターを選択します。

2. Topics メニューをクリックします。All topics ページ が表示されます。

3. トピックを選択します。

   

   トピックの概要ページが表示されます。

4. Schema タブをクリックします。

   デフォルトでは、Value スキーマが表示されます。

   

   ちなみに

   Control Center スキーマの表示では、スキーマに関するメタ情報の配置順序を変更できます。このビューと前のセクションに掲載されている例を比較すると、スキーマ名、名前空間、レコードタイプがフィールド定義の下に表示されていることがわかります。これは Control Center 内での表示を操作したためであり、スキーマ定義は同じです。

5. キースキーマが存在する場合は、Key タブをクリックして表示します。

Control Center でのスキーマの編集

トピックの既存のスキーマを編集します。

1. ナビゲーションバーでクラスターを選択します。

2. Topics メニューをクリックします。All topics ページ が表示されます。

3. トピックを選択します。

4. Schema タブをクリックします。

5. スキーマの Value または Key タブを選択します。

6. スキーマの任意の場所をクリックし、編集モードを有効にして、スキーマエディターで変更を加えます。

   たとえば、以下の例に倣っているとします。

   • my-transactions トピックを選択し、Schema をクリックして Value タブを選択します。

   • 新しい region フィールド用の以下の定義を、id および amount のフィールドの後にコピーアンドペーストしてスキーマを編集します。構文に従って、新しい定義の前にコンマを付けます。

     {
      "name": "region",
      "type": "string",
      "default": ""
     }
     

     Copy

   新しい region フィールドにデフォルト値が含まれていることにより、後方互換性が実現されています。デフォルト値をプラグすることにより、コンシューマーは、従来のスキーマ（region フィールドのないスキーマ）を使うプロデューサーが送信したデータを、この新しいスキーマを使用して読み取ることができます。

7. Save をクリックします。

   • スキーマのアップデートが有効であり、前のバージョンとの互換性がある場合は（後方互換モードを想定）、スキーマがアップデートされ、バージョンのカウントが増えます。スキーマを バージョン間で比較 することができます。

     

   • スキーマのアップデートが無効であったり、前のスキーマのバージョンとの互換性がなかったりする場合は、エラーが表示されます。

     以下の例は、別の新しいフィールド country の追加を示しています。後方互換性を確保するためのデフォルトは規定されていません。

     

Control Center でのスキーマのバージョンの比較

スキーマのバージョンを比較して、その進化の違いを表示します。

1. ナビゲーションバーでクラスターを選択します。

2. Topics メニューをクリックします。All topics ページ が表示されます。

3. トピックを選択します。

4. Schema タブをクリックします。

5. スキーマの Key または Value タブを選択します。

6. インラインメニューから Version history を選択します。

   

   スキーマの現在のバージョン番号がバージョンメニューに示されています。

   

7. Turn on version diff チェックボックスをオンにします。

8. 比較するバージョンを、各バージョンメニューから選択します。比較の差異が強調表示されます。

   

Control Center でのスキーマの互換モードの変更

デフォルトの互換モードは Backward です。トピックのスキーマのモードはすべて、必要に応じて変更することができます。

注意

既に本稼働環境で使用されている既存のスキーマの互換性モードを変更すると、アプリケーションの互換性が破られる可能性があるのでご注意ください。

1. ナビゲーションバーでクラスターを選択します。

2. Topics メニューをクリックします。All topics ページ が表示されます。

3. トピックを選択します。

4. Schema タブをクリックします。

5. スキーマの Key または Value タブを選択します。

6. インラインメニューから Compatibility setting を選択します。

   Compatibility settings が表示されます。

   

7. モードのオプションを選択します。

   • Backward （Confluent Schema Registry のデフォルト）
   • Transitive backward
   • Forward
   • Transitive forward
   • Full
   • Transitive full
   • None （非推奨）

   説明には、各オプションの互換性の動作が示されています。各オプションで許可されている変更などの詳細については、「スキーマ進化と互換性」を参照してください。

8. Save をクリックします。

Control Center からのスキーマのダウンロード

1. ナビゲーションバーでクラスターを選択します。

2. Topics メニューをクリックします。All topics ページ が表示されます。

3. トピックを選択します。

4. Schema タブをクリックします。

5. スキーマの Key または Value タブを選択します。

6. Download をクリックします。トピックのスキーマの JSON ファイルは、Downloads ディレクトリにダウンロードされます。

   ファイル名の例: schema-transactions-v1-Ry_XaOGvTxiZVZ5hbBhWRA.json

   内容の例:

   {"subject":"transactions-value","version":1,"id":2,"schema":"{\"type\":\"record\",\"name\":\"Payment\",
   \"namespace\":\"io.confluent.examples.clients.basicavro\",
   \"fields\":[{\"name\":\"id\",\"type\":\"string\"},{\"name\":\"amount\",\"type\":\"double\"},
   {\"name\":\"region\",\"type\":\"string\"}]}"}
   

   Copy

   これはスキーマの最初のバージョンであり、id は 2 です。スキーマは、エスケープされた JSON です。二重引用符の前には、バックスラッシュが付けられています。

"Schema Registry is not set up" エラーのトラブルシューティング

トピックのスキーマにアクセスしようとしたときに Control Center でエラーメッセージ（"Schema Registry is not set up"）が表示される場合は、まず Schema Registry が実行されていることを確認します。次に、Schema Registry listeners 構成が Control Center の confluent.controlcenter.schema.registry.url 構成と一致していることを確認します。HTTPS 構成パラメーターもチェックします。

詳細については、「メッセージ値用のスキーマの 1 つがこのトピック用に設定されていない」と、Confluent Platform を実行するために使用している方法に応じて「Confluent Platform Community コンポーネントを使用した Apache Kafka のクイックスタート（ローカル）」、「Confluent Platform Community コンポーネントを使用した Apache Kafka のクイックスタート（Docker）」、「オンプレミスのデプロイ」の起動手順を参照してください。

Control Center 内で Schema Registry を有効および無効にする

Control Center 内でスキーマを操作できる機能は、デフォルトで有効になります。ユーザーにこの機能を利用させない組織では、この機能を無効にできます。この機能を無効にした後は、Control Center UI に Topics Schema メニューおよび Schema タブは表示されなくなります。スキーマを表示および編集する機能は無効になります。

Control Center 内でスキーマの編集機能を無効にするには、以下の手順に従います。

1. ご使用の control-center.properties ファイル内で confluent.controlcenter.schema.registry.enable オプションを false に設定します。

   confluent.controlcenter.schema.registry.enable=false
   

   Copy

   注釈

   ご使用の環境向けに構成されている適切な Control Center プロパティファイル 1 つまたは複数に変更を加えます。対象のプロパティファイルには control-center-dev.properties または control-center-production.properties があります。プロパティファイルは /path-to-confluent/etc/confluent-control-center/ にあります。

2. Control Center を再起動し、プロパティファイルを渡して構成を有効にします。

   ./bin/control-center-stop
   ./bin/control-center-start ../etc/confluent-control-center/control-center.properties
   

   Copy

   ちなみに

   Confluent Platform 開発環境で confluent local を使用している場合は、以下のように停止および起動します。

   |confluent_local_stop_control_center|
   |confluent_local_stop_control_center| ../etc/confluent-control-center/control-center-dev.properties
   

   Copy

この機能を再度有効にするには、このオプションを true に設定し直して、アップデートしたプロパティファイルを使用して Control Center を再起動します。

マルチクラスター Schema Registry の有効化

バージョン 5.4.1 以降の Confluent Platform では、複数のスキーマレジストリを実行し、一意の Schema Registry をマルチクラスター環境内の各 Kafka クラスターに関連付ける機能をサポートしています。

Kafka クラスターと連携してスキーマレジストリをスケールアップする機能は、ビジネスの進化に役立ち、特に大企業のデータガバナンス、編成、部門横断的な管理をサポートします。

マルチクラスター Schema Registry が構成されて実行されている場合、通常どおりに Control Center 内でトピックごとにスキーマを作成および管理できます。

構成のプロパティおよびファイル

複数の Schema Registry クラスターは、該当する Control Center プロパティファイル内で、confluent.controlcenter.schema.registry.{name}.url を使用して指定できます。この方法で識別された Schema Registry クラスターを使用するには、以下のブローカーおよび Control Center の構成を追加するか確認します。

新しいエンドポイント /v1/metadata/schemaRegistryUrls が Kafka によって公開されて、confluent.schema.registry.url フィールドを Kafka ブローカー構成から返すようになっています。Control Center ではこのフィールドを使用して Kafka ブローカー構成からレジストリを検索します。これを使用するには、クラスターごとに一意にリスナーエンドポイントを構成する必要があります。

• ブローカーの server.properties ファイル（Kafka クラスターごとに一意）内で、confluent.http.server.listeners フィールドを使用して REST エンドポイントを指定します。デフォルトは http://0.0.0.0:8090 です。
• 該当する Control Center プロパティファイルで、confluent.controlcenter.streams.cprest.url を使用して controlcenter.cluster の REST エンドポイントを定義します。
• 追加のクラスターについては、confluent.controlcenter.kafka.{name}.cprest.url を使用して REST エンドポイントを定義します。これは、confluent.controlcenter.kafka.{name}.bootstrap.servers など、他の Kafka Control Center 構成で使用される Kafka クラスター名と一致している必要があります。

実行可能な最小構成では以下のファイルを編集し、以下のプロパティの設定を含みます（示されている名前とポートは例です）。

Control Center プロパティファイル

Control Center のプロパティファイル には、以下のプロパティが含まれます。

• confluent.controlcenter.schema.registry.url=http://localhost:8081
• confluent.controlcenter.schema.registry.sr-1.url=http://localhost:8082
• confluent.controlcenter.streams.cprest.url=http://localhost:8090
• confluent.controlcenter.kafka.AK1.cprest.url=http://localhost:8091

confluent.controlcenter.schema.registry.url の詳細については「Control Center の構成リファレンス」を参照してください。

Control Center クラスター用のブローカー構成ファイル

server0.properties など、controlcenter.cluster 用の Kafka ブローカー構成ファイルには、以下のプロパティが含まれます。

• confluent.http.server.listeners=http://localhost:8090
• confluent.schema.registry.url=http://localhost:8081

Kafka クラスター用のブローカー構成ファイル

AK1 用の Kafka ブローカー構成ファイル server1.properties には、以下のプロパティが含まれます。

• confluent.http.server.listeners=http://localhost:8091
• confluent.schema.registry.url=http://localhost:8082

これらの構成を使い、Control Center UI からスキーマを編集すると controlcenter.cluster 用の http://localhost:8081 への接続と AK1 用の http://localhost:8082 への接続が行われることになります。

デフォルトおよびフォールバック

confluent.schema.registry.url フィールドが指定されていないブローカーがある場合は、confluent.controlcenter.schema.registry.url Schema Registry URL が適用されます。たとえば、AK1 用に Schema Registry URL が指定されなかった場合、AK1 の関連付けられた Schema Registry クラスターも http://localhost:8081 で指定されます。Control Center プロパティファイル内で、confluent.controlcenter.schema.registry.url が明示的に指定されていない場合は、デフォルトの http://localhost:8081 が使用されます。

サンプル

以下の詳細な例には、機能するマルチクラスター Schema Registry セットアップが示されており、Control Center に接続された 2 つの Kafka クラスターが controlcenter.cluster および AK1 という名前を持ち、それぞれに 1 つのブローカーがあります。

以下のテーブルに 2 つのクラスター例それぞれのファイルおよび構成をまとめてあります。これらは完全なプロパティファイルではなく各ファイルからの抜粋であり、このセットアップ例で重要であるか、一般にデフォルトのプロパティファイルには含まれていない、マルチクラスター Schema Registry の仕様、ポート番号、および構成を定義しています。

この例を実行するには、デフォルトの Apache ZooKeeper™、Kafka （server.properties）、Schema Registry、および Control Center のプロパティファイルを、以下のファイル名の例に従って新しいファイルにコピーし、示されているようにプロパティの追加と変更を行ってから、「サンプルの実行」の説明に従ってコンポーネントを実行します。

重要な構成

ファイル	プロパティ
zookeeper0.properties	dataDir=/tmp/zookeeper/zk-0（未使用のログ用ディレクトリ） clientPort=2181（zookeeper.properties と同じ）
zookeeper1.properties	dataDir=/tmp/zookeeper/zk-1（未使用のログ用ディレクトリ） clientPort=2182（zookeeper0 が既に 2181 を使用しているため）
server0.properties	broker.id=0 listeners=PLAINTEXT://:9092 log.dirs=/tmp/kafka-logs-bk-0（未使用のログ用ディレクトリ） zookeeper.connect=localhost:2181 confluent.metrics.reporter.bootstrap.servers=localhost:9092 次の構成は、このブローカーのマルチクラスター |sr| セットアップに固有です。 confluent.http.server.listeners=http://0.0.0.0:8090 confluent.schema.registry.url=http://localhost:8081
server1.properties	broker.id=1 listeners=PLAINTEXT://:9093 log.dirs=/tmp/kafka-logs-bk-1（未使用のログ用ディレクトリ） zookeeper.connect=localhost:2182 confluent.metrics.reporter.bootstrap.servers=localhost:9093 次の構成は、このブローカーのマルチクラスター Schema Registry セットアップに固有です。 confluent.http.server.listeners=http://0.0.0.0:8091 confluent.schema.registry.url=http://localhost:8082
schema-registry0.properties	listeners=http://0.0.0.0:8081 kafkastore.connection.url=localhost:2181
schema-registry1.properties	listeners=http://0.0.0.0:8082 kafkastore.connection.url=localhost:2182
control-center-multi-sr.properties	bootstrap.servers=localhost:9092 zookeeper.connect=localhost:2181 confluent.controlcenter.kafka.AK1.bootstrap.servers=localhost:9093 confluent.controlcenter.kafka.AK1.zookeeper.connect=localhost:2182 confluent.controlcenter.streams.cprest.url=http://0.0.0.0:8090 confluent.controlcenter.schema.registry.url=http://localhost:8081 confluent.controlcenter.kafka.AK1.cprest.url=http://0.0.0.0:8091 confluent.controlcenter.schema.registry.SR-AK1.url=http://localhost:8082 cpcrest.url、confluent.controlcenter.kafka.AK1.cprest.url、confluent.controlcenter.schema.registry.SR-AK1.url の構成は、マルチクラスター Schema Registry に固有の新規プロパティです。

サンプルの実行

サンプルを実行するには、以下の手順に従います。

1. ZooKeeper を起動します。
2. Kafka ブローカーを起動します
3. Schema Registry クラスターを起動します。
4. Confluent Control Center を起動します。

以下の起動コマンドの例では、プロパティファイルが以下に示す <path-to-confluent>/etc/ ディレクトリにあり、コマンドは <path-to-confluent> から実行していることを前提としています。

ZooKeeper の起動

./bin/zookeeper-server-start etc/kafka/zookeeper0.properties

Copy

./bin/zookeeper-server-start etc/kafka/zookeeper1.properties

Copy

Kafka ブローカーの起動

./bin/kafka-server-start etc/kafka/server0.properties

Copy

./bin/kafka-server-start etc/kafka/server1.properties

Copy

スキーマレジストリクラスターの起動

./bin/schema-registry-start etc/schema-registry/schema-registry0.properties

Copy

./bin/schema-registry-start etc/schema-registry/schema-registry1.properties

Copy

Control Center の起動

./bin/control-center-start etc/confluent-control-center/control-center-multi-sr.properties

Copy

Control Center 上で両方のクラスターのスキーマを管理する

1. サンプルクラスターが実行中になり、Control Center による初期化が終了したら、ウェブブラウザー内で Control Center を開きます（Control Center は「Control Center ユーザーガイド」に記載されているように、デフォルトでは http://localhost:9021/ で実行されます）。
2. ナビゲーションバーからクラスターを選択し、Topics メニューをクリックして、「Control Center でのトピックのスキーマの作成」から開始して、1 つまたは両方のクラスターのスキーマ管理オプションを調べます。

Security

Control Center を使用して Schema Registry クライアントをセットアップするために使用されるその他のすべての構成は、confluent.controlcenter.schema.registry プレフィックスに Schema Registry クラスターの名前を追加するだけで、追加の Schema Registry クラスター用に構成できます。

たとえば、マルチクラスター Schema Registry を使用する HTTP 基本認証では、Confluent Control Center 構成ファイル内に以下の内容を指定します。

• confluent.controlcenter.schema.registry.basic.auth.credentials.source および confluent.controlcenter.schema.registry.basic.auth.user.info を使用して confluent.controlcenter.schema.registry.url クラスター用の認証を定義します。
• 追加の Schema Registry クラスター用には confluent.controlcenter.schema.registry.{name}.basic.auth.credentials.source および confluent.controlcenter.schema.registry.{name}.basic.auth.user.info を使用します（{name} によって URL のフィールドと関連付け）。

一部の Schema Registry クライアント構成には、schema.registry プレフィックスも含まれています。SSL セキュリティ設定のためには、Confluent Control Center 構成ファイル内に以下の内容を指定します。

• confluent.controlcenter.schema.registry.url クラスター用には confluent.controlcenter.schema.registry.schema.registry.ssl.truststore.location および confluent.controlcenter.schema.registry.schema.registry.ssl.truststore.password を使用します。
• 追加の Schema Registry クラスター用には confluent.controlcenter.schema.registry.{name}.schema.registry.ssl.truststore.location および confluent.controlcenter.schema.registry.{name}.schema.registry.ssl.truststore.password を使用します（{name} によって URL のフィールドと関連付け）。

詳細については、「Control Center 構成リファレンス」の「Kafka の暗号化、認証、認可設定」にある Schema Registry 認証プロパティ、および「HTTP 基本認証」の「Schema Registry」のセクションを参照してください。「Schema Registry セキュリティの概要」の「Schema Registry へのクライアントの構成方法」も参照してください。

エラーとトラブルシューティング

クラスターのブローカーに対応する Schema Registry URL があっても、該当する URL が Control Center プロパティファイル内に定義されていない場合は、クラスターの概要ページにエラーメッセージが表示されます。

バージョンの互換性

Control Center プロパティファイル内の confluent.controlcenter.schema.registry.url は、クラスターのブローカー構成に confluent.schema.registry.url フィールドが含まれていない場合のデフォルトになります。複数の Schema Registry クラスターは、confluent.controlcenter.schema.registry.{name}.url フィールドを使用して指定できます。

マルチクラスター Schema Registry は、Confluent Platform 5.4.0 で使用する現行バージョンの Kafka 2.4.x より前の Kafka バージョンで使用できません。ただし、単一クラスター Schema Registry セットアップを使用する場合は、前の Kafka バージョンで使用できます。詳細については「Confluent Platform および Apache Kafka の互換性」を参照してください。

おすすめの関連情報

• スキーマの開発、トピックへのマッピング、メッセージの送信に関する実践的な例については、「Schema Registry のチュートリアル」を参照してください。
• Schema Registry およびスキーマ管理の概要については、「Schema Registry の概要」を参照してください。
• Control Center 構成リファレンス
• Control Center のプロパティファイル
• 「Control Center のトラブルシューティング」の「メッセージ値用のスキーマの 1 つがこのトピック用に設定されていない」
• Confluent Server 上の Schema Validation
• トピックのサブジェクト命名方法を変更する