サードパーティー製のスキーマレジストリから AWS Glue Schema Registry への移行

サードパーティー製スキーマレジストリから AWS Glue Schema Registry への移行作業は、その時点での既存のサードパーティー製スキーマレジストリにより異なります。サードパーティーのスキーマレジストリを使用して送信された Apache Kafka トピック内にレコードがある場合、コンシューマはサードパーティーのスキーマレジストリを使用して、それらのレコードを非シリアル化する必要があります。AWSKafkaAvroDeserializer には、セカンダリのデシリアライザクラスを定義する機能があります。このクラスは、サードパーティーのデシリアライザを特定し、対象のレコードを非シリアル化するために使用できます、

サードパーティー製スキーマには、使用停止に関して 2 つの基準があります。1 つ目の基準では、サードパーティー製スキーマレジストリを使用する Apache Kafka トピックのレコードを必要とするコンシューマが、なくなった後にのみ使用停止となります。2 つ目の基準では、トピックに指定された保持期間に応じ、Apache Kafka トピックの試用期間が終了することで使用停止となります。無制限の保持期間を持つトピックの場合は、AWS Glue Schema Registry への移行は可能ですが、サードパーティー製スキーマレジストリを使用停止にすることはできません。この回避策としては、アプリケーションまたは Mirror Maker 2 により現在のトピックを読み取り、AWS Glue Schema Registry を使用する新しいトピックとして作成し直すことができます。

サードパーティー製スキーマレジストリから AWS Glue Schema Registry へ移行するには

1. AWS Glue Schema Registry にレジストリを作成するか、デフォルトのレジストリを使用します。

2. コンシューマを停止します。AWS Glue Schema Registry をプライマリのデシリアライザとして含めるように、コンシューマを変更します、サードパーティーのスキーマレジストリをセカンダリに変更します。

   • コンシューマのプロパティを設定します。この例では、secondary_decerializer は別のデシリアライザに設定されています。動作は次のとおりです。コンシューマは Amazon MSK からレコードを取得し、最初に AWSKafkaAvroDeserializer の使用を試みます。AWS Glue Schema Registry のスキーマの、Avro スキーマ ID を含むマジックバイトを読み取ることができない場合、AWSKafkaAvroDeserializerは、secondary_deserializer で指定されるデシリアライザクラスの使用を試みます。セカンダリのデシリアライザに固有のプロパティは、以下に示すように、schema_registry_url_config や specific_avro_reader_config などのコンシューマプロパティでも指定する必要があります。

     consumerProps.setProperty(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
     consumerProps.setProperty(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, AWSKafkaAvroDeserializer.class.getName());
     consumerProps.setProperty(AWSSchemaRegistryConstants.AWS_REGION, KafkaClickstreamConsumer.gsrRegion);
     consumerProps.setProperty(AWSSchemaRegistryConstants.SECONDARY_DESERIALIZER, KafkaAvroDeserializer.class.getName());
     consumerProps.setProperty(KafkaAvroDeserializerConfig.SCHEMA_REGISTRY_URL_CONFIG, "URL for third-party schema registry");
     consumerProps.setProperty(KafkaAvroDeserializerConfig.SPECIFIC_AVRO_READER_CONFIG, "true");
     			

3. コンシューマを再起動します。

4. プロデューサを停止し、そのプロデューサで AWS Glue Schema Registry を指定します。

   1. プロデューサのプロパティを設定します。この例のプロデューサでは、デフォルトのレジストリと自動登録スキーマバージョンを使用しています。

      producerProps.setProperty(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
      producerProps.setProperty(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, AWSKafkaAvroSerializer.class.getName());
      producerProps.setProperty(AWSSchemaRegistryConstants.AWS_REGION, "us-east-2");
      producerProps.setProperty(AWSSchemaRegistryConstants.AVRO_RECORD_TYPE, AvroRecordType.SPECIFIC_RECORD.getName());
      producerProps.setProperty(AWSSchemaRegistryConstants.SCHEMA_AUTO_REGISTRATION_SETTING, "true");

5. (オプション) 既存のスキーマおよびスキーマバージョンを、現在のサードパーティー製スキーマレジストリから AWS Glue Schema Registry (AWS Glue Schema Registry のデフォルトレジストリ、または AWS Glue Schema Registry 内の特定の非デフォルトレジストリ) に手動で移行します。これには、サードパーティーのスキーマレジストリから JSON 形式でスキーマをエクスポートし、AWS Management Console または AWS CLI を使用しながら AWS Glue Schema Registry の新しいスキーマを作成します。

   このステップは、AWS CLI および AWS Management Console を使用して、新しく作成されたスキーマバージョンについて、以前のスキーマバージョンとの互換性チェックを有効にする必要がある場合、またはスキーマバージョンの自動登録を有効にした新しいスキーマで、プロデユーサがメッセージを送信する場合に重要です。

6. プロデューサを起動します。