スキーマレジストリの仕組み

スキーマの登録: スキーマがレジストリにまだ存在しない場合、スキーマを保存先の名前と同じスキーマ名で登録することができます (test_topic、test_stream、prod_firehose など)。あるいは、プロデューサーがスキーマのカスタム名を指定することも可能です。プロデューサーは、source: msk_kafka_topic_A などのメタデータとしてスキーマにキーと値のペアを追加したり、スキーマ作成時に AWS タグをそのスキーマに追加したりできます。スキーマが登録されると、Schema Registry はスキーマのバージョン ID をシリアライザに返します。スキーマが存在するものの、シリアライザが存在しない新しいバージョンを使用している場合、それを新しいバージョンとして登録する前に、Schema Registry はスキーマ参照で互換性ルールを確認して、新しいバージョンに関する互換性を確保します。

スキーマの登録には、手動登録と自動登録の 2 つの方法があります。スキーマを手動で登録するには、AWS Glue コンソールまたは CLI/SDK を使用します。

シリアライザ設定で自動登録が有効化されている場合は、スキーマの自動登録が実行されます。プロデューサ設定で REGISTRY_NAME が指定されていない場合、自動登録はデフォルトのレジストリ (default-registry) の下に新しいスキーマバージョンを登録します。自動登録プロパティの指定については、「SerDe ライブラリのインストール」を参照してください。

シリアライザがスキーマに関してデータレコードを検証: データを生成するアプリケーションがそのスキーマを登録した場合、Schema Registry のシリアライザは、アプリケーションによって生成されるレコードが、登録されたスキーマに一致するフィールドとデータ型で構造化されていることを検証します。レコードのスキーマが登録されたスキーマと一致しない場合、シリアライザは例外を出力し、アプリケーションは、そのレコードを送信先にデリバリすることに失敗します。

スキーマが存在せず、プロデューサー設定を介してスキーマ名が指定されてもいない場合は、スキーマがトピック名 (Apache Kafka または Amazon MSK の場合) あるいはストリーム名 (Kinesis Data Streams の場合) と同じ名前で作成されます。

すべてのレコードは、スキーマ定義とデータを含みます。スキーマ定義は、Schema Registry 内の既存のスキーマおよびバージョンに関して照会されます。

デフォルトでは、プロデューサは、登録済みスキーマのスキーマ定義とスキーマバージョン ID をキャッシュします。レコードのスキーマバージョン定義がキャッシュ内に保存されたものと一致しない場合、プロデューサーは Schema Registry を使用してスキーマの検証を試みます。スキーマのバージョンが有効と認められた場合、そのバージョン ID と定義はプロデューサにローカルにキャッシュされます。

SerDe ライブラリのインストール のステップ 3 のように、オプションのプロデューサプロパティにより、デフォルトのキャッシュ期間 (24 時間) を調整することが可能です。

レコードのシリアル化と配信: レコードがスキーマに準拠している場合、シリアライザは各レコードをスキーマのバージョン ID で装飾し、選択したデータ形式 (AVRO、JSON、Protobuf 他の形式は追加予定) に基づいてシリアル化した上で、そのレコードを (プロデューサーのオプション設定により) 圧縮し送信先に送ります。

コンシューマーがデータを非シリアル化: このデータを読み取るコンシューマーは、レコードペイロードからのスキーマバージョン ID を解析する、Schema Registry のデシリアライザ用ライブラリを使用します。

デシリアライザが Schema Registry からのスキーマをリクエスト: デシリアライザが特定のスキーマバージョン ID を持つレコードを初めて認識すると、そのデシリアライザは、スキーマバージョン ID を使用してスキーマレジストリからスキーマを要求します。その後、そのスキーマをコンシューマ上でローカルにキャッシュします。Schema Registry でレコードの非シリアル化が不可能な場合、コンシューマはレコードからのデータをログに記録し、処理を続行するかアプリケーションを停止できます。

デシリアライザがスキーマを使用してレコードを非シリアル化: デシリアライザは、Schema Registry からスキーマバージョン ID を取得する際に (プロデューサーによって送信されたレコードが圧縮されている場合は) レコードを解凍し、スキーマを使用してレコードを非シリアル化します。これで、アプリケーションはレコードを処理できます。