# AWS Glue Schema registry
**Note**
AWS Glue Schema Registry is not supported in the following Regions in the AWS Glue console: Middle East (UAE).
The AWS Glue Schema registry allows you to centrally discover, control, and evolve data stream schemas. A *schema* defines the structure and format of a data record. With AWS Glue Schema registry, you can manage and enforce schemas on your data streaming applications using convenient integrations with Apache Kafka, [Amazon Managed Streaming for Apache Kafka](https://aws.amazon.com/msk/), [Amazon Kinesis Data Streams](https://aws.amazon.com/kinesis/data-streams/), [Amazon Managed Service for Apache Flink](https://aws.amazon.com/kinesis/data-analytics/), and [AWS Lambda](https://aws.amazon.com/lambda/).
The Schema registry supports AVRO (v1.11.4) data format, JSON Data format with [JSON Schema format](https://json-schema.org/) for the schema (specifications Draft-04, Draft-06, and Draft-07) with JSON schema validation using the [Everit library](https://github.com/everit-org/json-schema), Protocol Buffers (Protobuf) versions proto2 and proto3 without support for `extensions` or `groups`, and Java language support, with other data formats and languages to come. Supported features include compatibility, schema sourcing via metadata, auto-registration of schemas, IAM compatibility, and optional ZLIB compression to reduce storage and data transfer. The Schema registry is serverless and free to use.
Using a schema as a data format contract between producers and consumers leads to improved data governance, higher quality data, and enables data consumers to be resilient to compatible upstream changes.
The Schema registry allows disparate systems to share a schema for serialization and de-serialization. For example, assume you have a producer and consumer of data. The producer knows the schema when it publishes the data. The Schema Registry supplies a serializer and deserializer for certain systems such as Amazon MSK or Apache Kafka.
For more information, see [How the schema registry works](schema-registry-works.md).
**Topics**
+ [Schemas](#schema-registry-schemas)
+ [Registries](#schema-registry-registries)
+ [Schema versioning and compatibility](#schema-registry-compatibility)
+ [Open source Serde libraries](#schema-registry-serde-libraries)
+ [Quotas of the Schema Registry](#schema-registry-quotas)
+ [How the schema registry works](schema-registry-works.md)
+ [Getting started with schema registry](schema-registry-gs.md)
## Schemas
A *schema* defines the structure and format of a data record. A schema is a versioned specification for reliable data publication, consumption, or storage.
In this example schema for Avro, the format and structure are defined by the layout and field names, and the format of the field names is defined by the data types (e.g., `string`, `int`).
```
{
"type": "record",
"namespace": "ABC_Organization",
"name": "Employee",
"fields": [
{
"name": "Name",
"type": "string"
},
{
"name": "Age",
"type": "int"
},
{
"name": "address",
"type": {
"type": "record",
"name": "addressRecord",
"fields": [
{
"name": "street",
"type": "string"
},
{
"name": "zipcode",
"type": "int"
}
]
}
}
]
}
```
In this example JSON schema draft-07 for JSON, the format is defined by the [JSON Schema organization](https://json-schema.org/).
```
{
"$id": "https://example.com/person.schema.json",
"$schema": "http://json-schema.org/draft-07/schema#",
"title": "Person",
"type": "object",
"properties": {
"firstName": {
"type": "string",
"description": "The person's first name."
},
"lastName": {
"type": "string",
"description": "The person's last name."
},
"age": {
"description": "Age in years which must be equal to or greater than zero.",
"type": "integer",
"minimum": 0
}
}
}
```
In this example for Protobuf, the format is defined by the [version 2 of the Protocol Buffers language (proto2)](https://developers.google.com/protocol-buffers/docs/reference/proto2-spec).
```
syntax = "proto2";
package tutorial;
option java_multiple_files = true;
option java_package = "com.example.tutorial.protos";
option java_outer_classname = "AddressBookProtos";
message Person {
optional string name = 1;
optional int32 id = 2;
optional string email = 3;
enum PhoneType {
MOBILE = 0;
HOME = 1;
WORK = 2;
}
message PhoneNumber {
optional string number = 1;
optional PhoneType type = 2 [default = HOME];
}
repeated PhoneNumber phones = 4;
}
message AddressBook {
repeated Person people = 1;
}
```
## Registries
A *registry* is a logical container of schemas. Registries allow you to organize your schemas, as well as manage access control for your applications. A registry has an Amazon Resource Name (ARN) to allow you to organize and set different access permissions to schema operations within the registry.
You may use the default registry or create as many new registries as necessary.
**AWS Glue Schema Registry Hierarchy**
| |
| --- |
| [\[See the AWS documentation website for more details\]](http://docs.aws.amazon.com/glue/latest/dg/schema-registry.html) |
| [\[See the AWS documentation website for more details\]](http://docs.aws.amazon.com/glue/latest/dg/schema-registry.html) |
| [\[See the AWS documentation website for more details\]](http://docs.aws.amazon.com/glue/latest/dg/schema-registry.html) |
| [\[See the AWS documentation website for more details\]](http://docs.aws.amazon.com/glue/latest/dg/schema-registry.html) |
## Schema versioning and compatibility
Each schema can have multiple versions. Versioning is governed by a compatibility rule that is applied on a schema. Requests to register new schema versions are checked against this rule by the Schema Registry before they can succeed.
A schema version that is marked as a checkpoint is used to determine the compatibility of registering new versions of a schema. When a schema first gets created the default checkpoint will be the first version. As the schema evolves with more versions, you can use the CLI/SDK to change the checkpoint to a version of a schema using the `UpdateSchema` API that adheres to a set of constraints. In the console, editing the schema definition or compatibility mode will change the checkpoint to the latest version by default.
Compatibility modes allow you to control how schemas can or cannot evolve over time. These modes form the contract between applications producing and consuming data. When a new version of a schema is submitted to the registry, the compatibility rule applied to the schema name is used to determine if the new version can be accepted. There are 8 compatibility modes: NONE, DISABLED, BACKWARD, BACKWARD\$1ALL, FORWARD, FORWARD\$1ALL, FULL, FULL\$1ALL.
In the Avro data format, fields may be optional or required. An optional field is one in which the `Type` includes null. Required fields do not have null as the `Type`.
In the Protobuf data format, fields can be optional (including repeated) or required in proto2 syntax, while all fields are optional (including repeated) in proto3 syntax. All compatibility rules are determined based on the understanding of the Protocol Buffers specifications as well as the guidance from the [Google Protocol Buffers documentation](https://developers.google.com/protocol-buffers/docs/overview#updating).
+ *NONE*: No compatibility mode applies. You can use this choice in development scenarios or if you do not know the compatibility modes that you want to apply to schemas. Any new version added will be accepted without undergoing a compatibility check.
+ *DISABLED*: This compatibility choice prevents versioning for a particular schema. No new versions can be added.
+ *BACKWARD*: This compatibility choice is recommended because it allows consumers to read both the current and the previous schema version. You can use this choice to check compatibility against the previous schema version when you delete fields or add optional fields. A typical use case for BACKWARD is when your application has been created for the most recent schema.
**AVRO**
For example, assume you have a schema defined by first name (required), last name (required), email (required), and phone number (optional).
If your next schema version removes the required email field, this would successfully register. BACKWARD compatibility requires consumers to be able to read the current and previous schema version. Your consumers will be able to read the new schema as the extra email field from old messages is ignored.
If you have a proposed new schema version that adds a required field, for example, zip code, this would not successfully register with BACKWARD compatibility. Your consumers on the new version would not be able to read old messages before the schema change, as they are missing the required zip code field. However, if the zip code field was set as optional in the new schema, then the proposed version would successfully register as consumers can read the old schema without the optional zip code field.
**JSON**
For example, assume you have a schema version defined by first name (optional), last name (optional), email (optional) and phone number (optional).
If your next schema version adds the optional phone number property, this would successfully register as long as the original schema version does not allow any additional properties by setting the `additionalProperties` field to false. BACKWARD compatibility requires consumers to be able to read the current and previous schema version. Your consumers will be able to read data produced with the original schema where phone number property does not exist.
If you have a proposed new schema version that adds the optional phone number property, this would not successfully register with BACKWARD compatibility when the original schema version sets the `additionalProperties` field to true, namely allowing any additional property. Your consumers on the new version would not be able to read old messages before the schema change, as they cannot read data with phone number property in a different type, for example string instead of number.
**PROTOBUF**
For example, assume you have a schema version defined by a Message `Person` with `first name` (required), `last name` (required), `email` (required), and `phone number` (optional) fields under proto2 syntax.
Similar to AVRO scenarios, if your next schema version removes the required `email` field, this would successfully register. BACKWARD compatibility requires consumers to be able to read the current and previous schema version. Your consumers will be able to read the new schema as the extra `email` field from old messages is ignored.
If you have a proposed new schema version that adds a required field, for example, `zip code`, this would not successfully register with BACKWARD compatibility. Your consumers on the new version would not be able to read old messages before the schema change, as they are missing the required `zip code` field. However, if the `zip code` field was set as optional in the new schema, then the proposed version would successfully register as consumers can read the old schema without the optional `zip code` field.
In case of a gRPC use case, adding new RPC service or RPC method is a backward compatible change. For example, assume you have a schema version defined by an RPC service `MyService` with two RPC methods `Foo` and `Bar`.
If your next schema version adds a new RPC method called `Baz`, this would successfully register. Your consumers will be able to read data produced with the original schema according to BACKWARD compatibility since the newly added RPC method `Baz` is optional.
If you have a proposed new schema version that removes the existing RPC method `Foo`, this would not successfully register with BACKWARD compatibility. Your consumers on the new version would not be able to read old messages before the schema change, as they cannot understand and read data with the non-existent RPC method `Foo` in a gRPC application.
+ *BACKWARD\$1ALL*: This compatibility choice allows consumers to read both the current and all previous schema versions. You can use this choice to check compatibility against all previous schema versions when you delete fields or add optional fields.
+ *FORWARD*: This compatibility choice allows consumers to read both the current and the subsequent schema versions, but not necessarily later versions. You can use this choice to check compatibility against the last schema version when you add fields or delete optional fields. A typical use case for FORWARD is when your application has been created for a previous schema and should be able to process a more recent schema.
**AVRO**
For example, assume you have a schema version defined by first name (required), last name (required), email (optional).
If you have a new schema version that adds a required field, e.g. phone number, this would successfully register. FORWARD compatibility requires consumers to be able to read data produced with the new schema by using the previous version.
If you have a proposed schema version that deletes the required first name field, this would not successfully register with FORWARD compatibility. Your consumers on the prior version would not be able to read the proposed schemas as they are missing the required first name field. However, if the first name field was originally optional, then the proposed new schema would successfully register as the consumers can read data based on the new schema that doesn’t have the optional first name field.
**JSON**
For example, assume you have a schema version defined by first name (optional), last name (optional), email (optional) and phone number (optional).
If you have a new schema version that removes the optional phone number property, this would successfully register as long as the new schema version does not allow any additional properties by setting the `additionalProperties` field to false. FORWARD compatibility requires consumers to be able to read data produced with the new schema by using the previous version.
If you have a proposed schema version that deletes the optional phone number property, this would not successfully register with FORWARD compatibility when the new schema version sets the `additionalProperties` field to true, namely allowing any additional property. Your consumers on the prior version would not be able to read the proposed schemas as they could have phone number property in a different type, for example string instead of number.
**PROTOBUF**
For example, assume you have a schema version defined by a Message `Person` with `first name` (required), `last name` (required), `email` (optional) fields under proto2 syntax.
Similar to AVRO scenarios, if you have a new schema version that adds a required field, e.g. `phone number`, this would successfully register. FORWARD compatibility requires consumers to be able to read data produced with the new schema by using the previous version.
If you have a proposed schema version that deletes the required `first name` field, this would not successfully register with FORWARD compatibility. Your consumers on the prior version would not be able to read the proposed schemas as they are missing the required `first name` field. However, if the `first name` field was originally optional, then the proposed new schema would successfully register as the consumers can read data based on the new schema that doesn’t have the optional `first name` field.
In case of a gRPC use case, removing an RPC service or RPC method is a forward-compatible change. For example, assume you have a schema version defined by an RPC service `MyService` with two RPC methods `Foo` and `Bar`.
If your next schema version deletes the existing RPC method named `Foo`, this would successfully register according to FORWARD compatibility as the consumers can read data produced with the new schema by using the previous version. If you have a proposed new schema version that adds an RPC method `Baz`, this would not successfully register with FORWARD compatibility. Your consumers on the prior version would not be able to read the proposed schemas as they are missing the RPC method `Baz`.
+ *FORWARD\$1ALL*: This compatibility choice allows consumers to read data written by producers of any new registered schema. You can use this choice when you need to add fields or delete optional fields, and check compatibility against all previous schema versions.
+ *FULL*: This compatibility choice allows consumers to read data written by producers using the previous or next version of the schema, but not earlier or later versions. You can use this choice to check compatibility against the last schema version when you add or remove optional fields.
+ *FULL\$1ALL*: This compatibility choice allows consumers to read data written by producers using all previous schema versions. You can use this choice to check compatibility against all previous schema versions when you add or remove optional fields.
## Open source Serde libraries
AWS provides open-source Serde libraries as a framework for serializing and deserializing data. The open source design of these libraries allows common open-source applications and frameworks to support these libraries in their projects.
For more details on how the Serde libraries work, see [How the schema registry works](schema-registry-works.md).
## Quotas of the Schema Registry
Quotas, also referred to as limits in AWS, are the maximum values for the resources, actions, and items in your AWS account. The following are soft limits for the Schema Registry in AWS Glue.
**Schema version metadata key-value pairs**
You can have up to 10 key-value pairs per SchemaVersion per AWS Region.
You can view or set the key-value metadata pairs using the [QuerySchemaVersionMetadata action (Python: query\$1schema\$1version\$1metadata)](aws-glue-api-schema-registry-api.md#aws-glue-api-schema-registry-api-QuerySchemaVersionMetadata) or [PutSchemaVersionMetadata action (Python: put\$1schema\$1version\$1metadata)](aws-glue-api-schema-registry-api.md#aws-glue-api-schema-registry-api-PutSchemaVersionMetadata) APIs.
The following are hard limits for the Schema Registry in AWS Glue.
**Registries**
You can have up to 100 registries per AWS Region for this account.
**SchemaVersion**
You can have up to 10000 schema versions per AWS Region for this account.
Each new schema creates a new schema version, so you can theoretically have up to 10000 schemas per account per region, if each schema has only one version.
**Schema payloads**
There is a size limit of 170KB for schema payloads.
# How the schema registry works
This section describes how the serialization and deserialization processes in schema registry work.
1. Register a schema: If the schema doesn’t already exist in the registry, the schema can be registered with a schema name equal to the name of the destination (e.g., test\$1topic, test\$1stream, prod\$1firehose) or the producer can provide a custom name for the schema. Producers can also add key-value pairs to the schema as metadata, such as source: msk\$1kafka\$1topic\$1A, or apply AWS tags to schemas on schema creation. Once a schema is registered the Schema Registry returns the schema version ID to the serializer. If the schema exists but the serializer is using a new version that doesn’t exist, the Schema Registry will check the schema reference a compatibility rule to ensure the new version is compatible before registering it as a new version.
There are two methods of registering a schema: manual registration and auto-registration. You can register a schema manually via the AWS Glue console or CLI/SDK.
When auto-registration is turned on in the serializer settings, automatic registration of the schema will be performed. If `REGISTRY_NAME` is not provided in the producer configurations, then auto-registration will register the new schema version under the default registry (default-registry). See [Installing SerDe Libraries](schema-registry-gs-serde.md) for information on specifying the auto-registration property.
1. Serializer validates data records against the schema: When the application producing data has registered its schema, the Schema Registry serializer validates the record being produced by the application is structured with the fields and data types matching a registered schema. If the schema of the record does not match a registered schema, the serializer will return an exception and the application will fail to deliver the record to the destination.
If no schema exists and if the schema name is not provided via the producer configurations, then the schema is created with the same name as the topic name (if Apache Kafka or Amazon MSK) or stream name (if Kinesis Data Streams).
Every record has a schema definition and data. The schema definition is queried against the existing schemas and versions in the Schema Registry.
By default, producers cache schema definitions and schema version IDs of registered schemas. If a record’s schema version definition does not match what’s available in cache, the producer will attempt to validate the schema with the Schema Registry. If the schema version is valid, then its version ID and definition will be cached locally on the producer.
You can adjust the default cache period (24 hours) within the optional producer properties in step \$13 of [Installing SerDe Libraries](schema-registry-gs-serde.md).
1. Serialize and deliver records: If the record complies with the schema, the serializer decorates each record with the schema version ID, serializes the record based on the data format selected (AVRO, JSON, Protobuf, or other formats coming soon), compresses the record (optional producer configuration), and delivers it to the destination.
1. Consumers deserialize the data: Consumers reading this data use the Schema Registry deserializer library that parses the schema version ID from the record payload.
1. Deserializer may request the schema from the Schema Registry: If this is the first time the deserializer has seen records with a particular schema version ID, using the schema version ID the deserializer will request the schema from the Schema Registry and cache the schema locally on the consumer. If the Schema Registry cannot deserialize the record, the consumer can log the data from the record and move on, or halt the application.
1. The deserializer uses the schema to deserialize the record: When the deserializer retrieves the schema version ID from the Schema Registry, the deserializer decompresses the record (if record sent by producer is compressed) and uses the schema to deserialize the record. The application now processes the record.
**Note**
Encryption: Your clients communicate with the Schema Registry via API calls which encrypt data in-transit using TLS encryption over HTTPS. Schemas stored in the Schema Registry are always encrypted at rest using a service-managed AWS Key Management Service (AWS KMS) key.
**Note**
User Authorization: The Schema Registry supports identity-based IAM policies.
# Getting started with schema registry
The following sections provide an overview and walk you through setting up and using Schema Registry. For information about schema registry concepts and components, see [AWS Glue Schema registry](schema-registry.md).
**Topics**
+ [Installing SerDe Libraries](schema-registry-gs-serde.md)
+ [Integrating with AWS Glue Schema Registry](schema-registry-integrations.md)
+ [Migration from a third-party schema registry to AWS Glue Schema Registry](schema-registry-integrations-migration.md)
# Installing SerDe Libraries
The SerDe libraries provide a framework for serializing and deserializing data.
You will install the open source serializer for your applications producing data (collectively the "serializers"). The serializer handles serialization, compression, and the interaction with the Schema Registry. The serializer automatically extracts the schema from a record being written to a Schema Registry compatible destination, such as Amazon MSK. Likewise, you will install the open source deserializer on your applications consuming data.
# Java Implementation
**Note**
Prerequisites: Before completing the following steps, you will need to have a Amazon Managed Streaming for Apache Kafka (Amazon MSK) or Apache Kafka cluster running. Your producers and consumers need to be running on Java 8 or above.
To install the libraries on producers and consumers:
1. Inside both the producers’ and consumers’ pom.xml files, add this dependency via the code below:
```
software.amazon.glueschema-registry-serde1.1.5
```
Alternatively, you can clone the [AWS Glue Schema Registry Github repository](https://github.com/awslabs/aws-glue-schema-registry).
1. Setup your producers with these required properties:
```
props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName()); // Can replace StringSerializer.class.getName()) with any other key serializer that you may use
props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, GlueSchemaRegistryKafkaSerializer.class.getName());
props.put(AWSSchemaRegistryConstants.AWS_REGION, "us-east-2");
properties.put(AWSSchemaRegistryConstants.DATA_FORMAT, "JSON"); // OR "AVRO"
```
If there are no existing schemas, then auto-registration needs to be turned on (next step). If you do have a schema that you would like to apply, then replace "my-schema" with your schema name. Also the "registry-name" has to be provided if schema auto-registration is off. If the schema is created under the "default-registry" then registry name can be omitted.
1. (Optional) Set any of these optional producer properties. For detailed property descriptions, see [the ReadMe file](https://github.com/awslabs/aws-glue-schema-registry/blob/master/README.md).
```
props.put(AWSSchemaRegistryConstants.SCHEMA_AUTO_REGISTRATION_SETTING, "true"); // If not passed, uses "false"
props.put(AWSSchemaRegistryConstants.SCHEMA_NAME, "my-schema"); // If not passed, uses transport name (topic name in case of Kafka, or stream name in case of Kinesis Data Streams)
props.put(AWSSchemaRegistryConstants.REGISTRY_NAME, "my-registry"); // If not passed, uses "default-registry"
props.put(AWSSchemaRegistryConstants.CACHE_TIME_TO_LIVE_MILLIS, "86400000"); // If not passed, uses 86400000 (24 Hours)
props.put(AWSSchemaRegistryConstants.CACHE_SIZE, "10"); // default value is 200
props.put(AWSSchemaRegistryConstants.COMPATIBILITY_SETTING, Compatibility.FULL); // Pass a compatibility mode. If not passed, uses Compatibility.BACKWARD
props.put(AWSSchemaRegistryConstants.DESCRIPTION, "This registry is used for several purposes."); // If not passed, constructs a description
props.put(AWSSchemaRegistryConstants.COMPRESSION_TYPE, AWSSchemaRegistryConstants.COMPRESSION.ZLIB); // If not passed, records are sent uncompressed
```
Auto-registration registers the schema version under the default registry ("default-registry"). If a `SCHEMA_NAME` is not specified in the previous step, then the topic name is inferred as `SCHEMA_NAME`.
See [Schema versioning and compatibility](schema-registry.md#schema-registry-compatibility) for more information on compatibility modes.
1. Setup your consumers with these required properties:
```
props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, GlueSchemaRegistryKafkaDeserializer.class.getName());
props.put(AWSSchemaRegistryConstants.AWS_REGION, "us-east-2"); // Pass an AWS Region
props.put(AWSSchemaRegistryConstants.AVRO_RECORD_TYPE, AvroRecordType.GENERIC_RECORD.getName()); // Only required for AVRO data format
```
1. (Optional) Set these optional consumer properties. For detailed property descriptions, see [the ReadMe file](https://github.com/awslabs/aws-glue-schema-registry/blob/master/README.md).
```
properties.put(AWSSchemaRegistryConstants.CACHE_TIME_TO_LIVE_MILLIS, "86400000"); // If not passed, uses 86400000
props.put(AWSSchemaRegistryConstants.CACHE_SIZE, "10"); // default value is 200
props.put(AWSSchemaRegistryConstants.SECONDARY_DESERIALIZER, "com.amazonaws.services.schemaregistry.deserializers.external.ThirdPartyDeserializer"); // For migration fall back scenario
```
# C\$1 Implementation
**Note**
Prerequisites: Before completing the following steps, you will need to have a Amazon Managed Streaming for Apache Kafka (Amazon MSK) or Apache Kafka cluster running. Your producers and consumers need to be running on .NET 8.0 or above.
## Installation
For C\$1 applications, install the AWS Glue Schema Registry SerDe NuGet package using one of the following methods:
**.NET CLI:**
Use the following command to install the package:
```
dotnet add package Aws.Glue.SchemaRegistry --version 1.0.0-
```
where `` could be `1.0.0-linux-x64`, `1.0.0-linux-musl-x64` or `1.0.0-linux-arm64`
**PackageReference (in your .csproj file):**
Add the following to your project file:
```
```
where `` could be `1.0.0-linux-x64`, `1.0.0-linux-musl-x64` or `1.0.0-linux-arm64`
## Configuration File Setup
Create a configuration properties file (e.g., `gsr-config.properties`) with the required settings:
**Minimal Configuration:**
The following shows a minimal configuration example:
```
region=us-east-1
registry.name=default-registry
dataFormat=AVRO
schemaAutoRegistrationEnabled=true
```
## Using C\$1 Glue Schema client library for Kafka SerDes
**Sample serializer usage:**
The following example shows how to use the serializer:
```
private static readonly string PROTOBUF_CONFIG_PATH = "";
var protobufSerializer = new GlueSchemaRegistryKafkaSerializer(PROTOBUF_CONFIG_PATH);
var serialized = protobufSerializer.Serialize(message, message.Descriptor.FullName);
// send serialized bytes to Kafka using producer.Produce(serialized)
```
**Sample deserializer usage:**
The following example shows how to use the deserializer:
```
private static readonly string PROTOBUF_CONFIG_PATH = "";
var dataConfig = new GlueSchemaRegistryDataFormatConfiguration(
new Dictionary
{
{
GlueSchemaRegistryConstants.ProtobufMessageDescriptor, message.Descriptor
}
}
);
var protobufDeserializer = new GlueSchemaRegistryKafkaDeserializer(PROTOBUF_CONFIG_PATH, dataConfig);
// read message from Kafka using serialized = consumer.Consume()
var deserializedObject = protobufDeserializer.Deserialize(message.Descriptor.FullName, serialized);
```
## Using C\$1 Glue Schema client library with KafkaFlow for SerDes
**Sample serializer usage:**
The following example shows how to configure KafkaFlow with the serializer:
```
services.AddKafka(kafka => kafka
.UseConsoleLog()
.AddCluster(cluster => cluster
.WithBrokers(new[] { "localhost:9092" })
.AddProducer(producer => producer
.DefaultTopic("customer-events")
.AddMiddlewares(m => m
.AddSerializer>(
() => new GlueSchemaRegistryKafkaFlowProtobufSerializer("config/gsr-config.properties")
)
)
)
)
);
```
**Sample deserializer usage:**
The following example shows how to configure KafkaFlow with the deserializer:
```
.AddConsumer(consumer => consumer
.Topic("customer-events")
.WithGroupId("customer-group")
.WithBufferSize(100)
.WithWorkersCount(10)
.AddMiddlewares(middlewares => middlewares
.AddDeserializer>(
() => new GlueSchemaRegistryKafkaFlowProtobufDeserializer("config/gsr-config.properties")
)
.AddTypedHandlers(h => h.AddHandler())
)
)
```
## Optional Producer Properties
You can extend your configuration file with additional optional properties:
```
# Auto-registration (if not passed, uses "false")
schemaAutoRegistrationEnabled=true
# Schema name (if not passed, uses topic name)
schema.name=my-schema
# Registry name (if not passed, uses "default-registry")
registry.name=my-registry
# Cache settings
cacheTimeToLiveMillis=86400000
cacheSize=200
# Compatibility mode (if not passed, uses BACKWARD)
compatibility=FULL
# Registry description
description=This registry is used for several purposes.
# Compression (if not passed, records are sent uncompressed)
compressionType=ZLIB
```
## Supported Data Formats
Both Java and C\$1 implementations support the same data formats:
+ *AVRO*: Apache Avro binary format
+ *JSON*: JSON Schema format
+ *PROTOBUF*: Protocol Buffers format
## Notes
+ To get started with the library, please visit [https://www.nuget.org/packages/AWS.Glue.SchemaRegistry](https://www.nuget.org/packages/AWS.Glue.SchemaRegistry)
+ Source code is available at: [https://github.com/awslabs/aws-glue-schema-registry](https://github.com/awslabs/aws-glue-schema-registry)
# Creating a registry
You may use the default registry or create as many new registries as necessary using the AWS Glue APIs or AWS Glue console.
**AWS Glue APIs**
You can use these steps to perform this task using the AWS Glue APIs.
To use the AWS CLI for the AWS Glue Schema Registry APIs, make sure to update your AWS CLI to the latest version.
To add a new registry, use the [CreateRegistry action (Python: create\$1registry)](aws-glue-api-schema-registry-api.md#aws-glue-api-schema-registry-api-CreateRegistry) API. Specify `RegistryName` as the name of the registry to be created, with a max length of 255, containing only letters, numbers, hyphens, underscores, dollar signs, or hash marks.
Specify a `Description` as a string not more than 2048 bytes long, matching the [ URI address multi-line string pattern](https://docs.aws.amazon.com/glue/latest/dg/aws-glue-api-common.html#aws-glue-api-common-_string-patterns).
Optionally, specify one or more `Tags` for your registry, as a map array of key-value pairs.
```
aws glue create-registry --registry-name registryName1 --description description
```
When your registry is created it is assigned an Amazon Resource Name (ARN), which you can view in the `RegistryArn` of the API response. Now that you've created a registry, create one or more schemas for that registry.
**AWS Glue console**
To add a new registry in the AWS Glue console:
1. Sign in to the AWS Management Console and open the AWS Glue console at [https://console.aws.amazon.com/glue/](https://console.aws.amazon.com/glue\).
1. In the navigation pane, under **Data catalog**, choose **Schema registries**.
1. Choose **Add registry**.
1. Enter a **Registry name** for the registry, consisting of letters, numbers, hyphens, or underscores. This name cannot be changed.
1. Enter a **Description** (optional) for the registry.
1. Optionally, apply one or more tags to your registry. Choose **Add new tag** and specify a **Tag key** and optionally a **Tag value**.
1. Choose **Add registry**.
![\[Example of a creating a registry.\]](http://docs.aws.amazon.com/glue/latest/dg/images/schema_reg_create_registry.png)
When your registry is created it is assigned an Amazon Resource Name (ARN), which you can view by choosing the registry from the list in **Schema registries**. Now that you've created a registry, create one or more schemas for that registry.
# Dealing with a specific record (JAVA POJO) for JSON
You can use a plain old Java object (POJO) and pass the object as a record. This is similar to the notion of a specific record in AVRO. The [mbknor-jackson-jsonschema](https://github.com/mbknor/mbknor-jackson-jsonSchema) can generate a JSON schema for the POJO passed. This library can also inject additional information in the JSON schema.
The AWS Glue Schema Registry library uses the injected "className" field in schema to provide a fully classified class name. The "className" field is used by the deserializer to deserialize into an object of that class.
```
Example class :
@JsonSchemaDescription("This is a car")
@JsonSchemaTitle("Simple Car Schema")
@Builder
@AllArgsConstructor
@EqualsAndHashCode
// Fully qualified class name to be added to an additionally injected property
// called className for deserializer to determine which class to deserialize
// the bytes into
@JsonSchemaInject(
strings = {@JsonSchemaString(path = "className",
value = "com.amazonaws.services.schemaregistry.integrationtests.generators.Car")}
)
// List of annotations to help infer JSON Schema are defined by https://github.com/mbknor/mbknor-jackson-jsonSchema
public class Car {
@JsonProperty(required = true)
private String make;
@JsonProperty(required = true)
private String model;
@JsonSchemaDefault("true")
@JsonProperty
public boolean used;
@JsonSchemaInject(ints = {@JsonSchemaInt(path = "multipleOf", value = 1000)})
@Max(200000)
@JsonProperty
private int miles;
@Min(2000)
@JsonProperty
private int year;
@JsonProperty
private Date purchaseDate;
@JsonProperty
@JsonFormat(shape = JsonFormat.Shape.NUMBER)
private Date listedDate;
@JsonProperty
private String[] owners;
@JsonProperty
private Collection serviceChecks;
// Empty constructor is required by Jackson to deserialize bytes
// into an Object of this class
public Car() {}
}
```
# Creating a schema
You can create a schema using the AWS Glue APIs or the AWS Glue console.
**AWS Glue APIs**
You can use these steps to perform this task using the AWS Glue APIs.
To add a new schema, use the [CreateSchema action (Python: create\$1schema)](aws-glue-api-schema-registry-api.md#aws-glue-api-schema-registry-api-CreateSchema) API.
Specify a `RegistryId` structure to indicate a registry for the schema. Or, omit the `RegistryId` to use the default registry.
Specify a `SchemaName` consisting of letters, numbers, hyphens, or underscores, and `DataFormat` as **AVRO** or **JSON**. `DataFormat` once set on a schema is not changeable.
Specify a `Compatibility` mode:
+ *Backward (recommended)* — Consumer can read both current and previous version.
+ *Backward all* — Consumer can read current and all previous versions.
+ *Forward* — Consumer can read both current and subsequent version.
+ *Forward all* — Consumer can read both current and all subsequent versions.
+ *Full* — Combination of Backward and Forward.
+ *Full all* — Combination of Backward all and Forward all.
+ *None* — No compatibility checks are performed.
+ *Disabled* — Prevent any versioning for this schema.
Optionally, specify `Tags` for your schema.
Specify a `SchemaDefinition` to define the schema in Avro, JSON, or Protobuf data format. See the examples.
For Avro data format:
```
aws glue create-schema --registry-id RegistryName="registryName1" --schema-name testschema --compatibility NONE --data-format AVRO --schema-definition "{\"type\": \"record\", \"name\": \"r1\", \"fields\": [ {\"name\": \"f1\", \"type\": \"int\"}, {\"name\": \"f2\", \"type\": \"string\"} ]}"
```
```
aws glue create-schema --registry-id RegistryArn="arn:aws:glue:us-east-2:901234567890:registry/registryName1" --schema-name testschema --compatibility NONE --data-format AVRO --schema-definition "{\"type\": \"record\", \"name\": \"r1\", \"fields\": [ {\"name\": \"f1\", \"type\": \"int\"}, {\"name\": \"f2\", \"type\": \"string\"} ]}"
```
For JSON data format:
```
aws glue create-schema --registry-id RegistryName="registryName" --schema-name testSchemaJson --compatibility NONE --data-format JSON --schema-definition "{\"$schema\": \"http://json-schema.org/draft-07/schema#\",\"type\":\"object\",\"properties\":{\"f1\":{\"type\":\"string\"}}}"
```
```
aws glue create-schema --registry-id RegistryArn="arn:aws:glue:us-east-2:901234567890:registry/registryName" --schema-name testSchemaJson --compatibility NONE --data-format JSON --schema-definition "{\"$schema\": \"http://json-schema.org/draft-07/schema#\",\"type\":\"object\",\"properties\":{\"f1\":{\"type\":\"string\"}}}"
```
For Protobuf data format:
```
aws glue create-schema --registry-id RegistryName="registryName" --schema-name testSchemaProtobuf --compatibility NONE --data-format PROTOBUF --schema-definition "syntax = \"proto2\";package org.test;message Basic { optional int32 basic = 1;}"
```
```
aws glue create-schema --registry-id RegistryArn="arn:aws:glue:us-east-2:901234567890:registry/registryName" --schema-name testSchemaProtobuf --compatibility NONE --data-format PROTOBUF --schema-definition "syntax = \"proto2\";package org.test;message Basic { optional int32 basic = 1;}"
```
**AWS Glue console**
To add a new schema using the AWS Glue console:
1. Sign in to the AWS Management Console and open the AWS Glue console at [https://console.aws.amazon.com/glue/](https://console.aws.amazon.com/glue\).
1. In the navigation pane, under **Data catalog**, choose **Schemas**.
1. Choose **Add schema**.
1. Enter a **Schema name**, consisting of letters, numbers, hyphens, underscores, dollar signs, or hashmarks. This name cannot be changed.
1. Choose the **Registry** where the schema will be stored from the drop-down menu. The parent registry cannot be changed post-creation.
1. Leave the **Data format** as *Apache Avro* or *JSON*. This format applies to all versions of this schema.
1. Choose a **Compatibility mode**.
+ *Backward (recommended)* — receiver can read both current and previous versions.
+ *Backward All* — receiver can read current and all previous versions.
+ *Forward* — sender can write both current and previous versions.
+ *Forward All* — sender can write current and all previous versions.
+ *Full* — combination of Backward and Forward.
+ *Full All* — combination of Backward All and Forward All.
+ *None* — no compatibility checks performed.
+ *Disabled* — prevent any versioning for this schema.
1. Enter an optional **Description** for the registry of up to 250 characters.
![\[Example of a creating a schema.\]](http://docs.aws.amazon.com/glue/latest/dg/images/schema_reg_create_schema.png)
1. Optionally, apply one or more tags to your schema. Choose **Add new tag** and specify a **Tag key** and optionally a **Tag value**.
1. In the **First schema version** box, enter or paste your initial schema. .
For Avro format, see [Working with Avro data format](#schema-registry-avro)
For JSON format, see [Working with JSON data format](#schema-registry-json)
1. Optionally, choose **Add metadata** to add version metadata to annotate or classify your schema version.
1. Choose **Create schema and version**.
![\[Example of a creating a schema.\]](http://docs.aws.amazon.com/glue/latest/dg/images/schema_reg_create_schema2.png)
The schema is created and appears in the list under **Schemas**.
## Working with Avro data format
Avro provides data serialization and data exchange services. Avro stores the data definition in JSON format making it easy to read and interpret. The data itself is stored in binary format.
For information on defining an Apache Avro schema, see the [Apache Avro specification](http://avro.apache.org/docs/current/spec.html).
## Working with JSON data format
Data can be serialized with JSON format. [JSON Schema format](https://json-schema.org/) defines the standard for JSON Schema format.
# Updating a schema or registry
Once created you can edit your schemas, schema versions, or registry.
## Updating a registry
You can update a registry using the AWS Glue APIs or the AWS Glue console. The name of an existing registry cannot be edited. You can edit the description for a registry.
**AWS Glue APIs**
To update an existing registry, use the [UpdateRegistry action (Python: update\$1registry)](aws-glue-api-schema-registry-api.md#aws-glue-api-schema-registry-api-UpdateRegistry) API.
Specify a `RegistryId` structure to indicate the registry that you want to update. Pass a `Description` to change the description for a registry.
```
aws glue update-registry --description updatedDescription --registry-id RegistryArn="arn:aws:glue:us-east-2:901234567890:registry/registryName1"
```
**AWS Glue console**
To update a registry using the AWS Glue console:
1. Sign in to the AWS Management Console and open the AWS Glue console at [https://console.aws.amazon.com/glue/](https://console.aws.amazon.com/glue\).
1. In the navigation pane, under **Data catalog**, choose **Schema registries**.
1. Choose a registry from the the list of registries, by checking its box.
1. In the **Action** menu, choose **Edit registry**.
# Updating a schema
You can update the description or compatibility setting for a schema.
To update an existing schema, use the [UpdateSchema action (Python: update\$1schema)](aws-glue-api-schema-registry-api.md#aws-glue-api-schema-registry-api-UpdateSchema) API.
Specify a `SchemaId` structure to indicate the schema that you want to update. One of `VersionNumber` or `Compatibility` has to be provided.
Code example 11:
```
aws glue update-schema --description testDescription --schema-id SchemaName="testSchema1",RegistryName="registryName1" --schema-version-number LatestVersion=true --compatibility NONE
```
```
aws glue update-schema --description testDescription --schema-id SchemaArn="arn:aws:glue:us-east-2:901234567890:schema/registryName1/testSchema1" --schema-version-number LatestVersion=true --compatibility NONE
```
# Adding a schema version
When you add a schema version, you will need to compare the versions to make sure the new schema will be accepted.
To add a new version to an existing schema, use the [RegisterSchemaVersion action (Python: register\$1schema\$1version)](aws-glue-api-schema-registry-api.md#aws-glue-api-schema-registry-api-RegisterSchemaVersion) API.
Specify a `SchemaId` structure to indicate the schema for which you want to add a version, and a `SchemaDefinition` to define the schema.
Code example 12:
```
aws glue register-schema-version --schema-definition "{\"type\": \"record\", \"name\": \"r1\", \"fields\": [ {\"name\": \"f1\", \"type\": \"int\"}, {\"name\": \"f2\", \"type\": \"string\"} ]}" --schema-id SchemaArn="arn:aws:glue:us-east-1:901234567890:schema/registryName/testschema"
```
```
aws glue register-schema-version --schema-definition "{\"type\": \"record\", \"name\": \"r1\", \"fields\": [ {\"name\": \"f1\", \"type\": \"int\"}, {\"name\": \"f2\", \"type\": \"string\"} ]}" --schema-id SchemaName="testschema",RegistryName="testregistry"
```
1. Sign in to the AWS Management Console and open the AWS Glue console at [https://console.aws.amazon.com/glue/](https://console.aws.amazon.com/glue\).
1. In the navigation pane, under **Data catalog**, choose **Schemas**.
1. Choose the schema from the the list of schemas, by checking its box.
1. Choose one or more schemas from the list, by checking the boxes.
1. In the **Action** menu, choose **Register new version**.
1. In the **New version** box, enter or paste your new schema.
1. Choose **Compare with previous version** to see differences with the previous schema version.
1. Optionally, choose **Add metadata** to add version metadata to annotate or classify your schema version. Enter **Key** and optional **Value**.
1. Choose **Register version**.
![\[Adding a schema version.\]](http://docs.aws.amazon.com/glue/latest/dg/images/schema_reg_add_schema_version.png)
The schema(s) version appears in the list of versions. If the version changed the compatibility mode, the version will be marked as a checkpoint.
## Example of a schema version comparison
When you choose to **Compare with previous version**, you will see the previous and new versions displayed together. Changed information will be highlighted as follows:
+ *Yellow*: indicates changed information.
+ *Green*: indicates content added in the latest version.
+ *Red*: indicates content removed in the latest version.
You can also compare against earlier versions.
![\[Example of a schema version comparison.\]](http://docs.aws.amazon.com/glue/latest/dg/images/schema_reg_version_comparison.png)
# Deleting a schema or registry
Deleting a schema, a schema version, or a registry are permanent actions that cannot be undone.
## Deleting a schema
You may want to delete a schema when it will no longer be used within a registry, using the AWS Management Console, or the [DeleteSchema action (Python: delete\$1schema)](aws-glue-api-schema-registry-api.md#aws-glue-api-schema-registry-api-DeleteSchema) API.
Deleting one or more schemas is a permanent action that cannot be undone. Make sure that the schema or schemas are no longer needed.
To delete a schema from the registry, call the [DeleteSchema action (Python: delete\$1schema)](aws-glue-api-schema-registry-api.md#aws-glue-api-schema-registry-api-DeleteSchema) API, specifying the `SchemaId` structure to identify the schema.
For example:
```
aws glue delete-schema --schema-id SchemaArn="arn:aws:glue:us-east-2:901234567890:schema/registryName1/schemaname"
```
```
aws glue delete-schema --schema-id SchemaName="TestSchema6-deleteschemabyname",RegistryName="default-registry"
```
**AWS Glue console**
To delete a schema from the AWS Glue console:
1. Sign in to the AWS Management Console and open the AWS Glue console at [https://console.aws.amazon.com/glue/](https://console.aws.amazon.com/glue\).
1. In the navigation pane, under **Data catalog**, choose **Schema registries**.
1. Choose the registry that contains your schema from the the list of registries.
1. Choose one or more schemas from the list, by checking the boxes.
1. In the **Action** menu, choose **Delete schema**.
1. Enter the text **Delete** in the field to confirm deletion.
1. Choose **Delete**.
The schema(s) you specified are deleted from the registry.
## Deleting a schema version
As schemas accumulate in the registry, you may want to delete unwanted schema versions using the AWS Management Console, or the [DeleteSchemaVersions action (Python: delete\$1schema\$1versions)](aws-glue-api-schema-registry-api.md#aws-glue-api-schema-registry-api-DeleteSchemaVersions) API. Deleting one or more schema versions is a permanent action that cannot be undone. Make sure that the schema versions are no longer needed.
When deleting schema versions, take note of the following constraints:
+ You cannot delete a check-pointed version.
+ The range of contiguous versions cannot be more than 25.
+ The latest schema version must not be in a pending state.
Specify the `SchemaId` structure to identify the schema, and specify `Versions` as a range of versions to delete. For more information on specifying a version or range of versions, see [DeleteRegistry action (Python: delete\$1registry)](aws-glue-api-schema-registry-api.md#aws-glue-api-schema-registry-api-DeleteRegistry). The schema versions you specified are deleted from the registry.
Calling the [ListSchemaVersions action (Python: list\$1schema\$1versions)](aws-glue-api-schema-registry-api.md#aws-glue-api-schema-registry-api-ListSchemaVersions) API after this call will list the status of the deleted versions.
For example:
```
aws glue delete-schema-versions --schema-id SchemaName="TestSchema6",RegistryName="default-registry" --versions "1-1"
```
```
aws glue delete-schema-versions --schema-id SchemaArn="arn:aws:glue:us-east-2:901234567890:schema/default-registry/TestSchema6-NON-Existent" --versions "1-1"
```
1. Sign in to the AWS Management Console and open the AWS Glue console at [https://console.aws.amazon.com/glue/](https://console.aws.amazon.com/glue\).
1. In the navigation pane, under **Data catalog**, choose **Schema registries**.
1. Choose the registry that contains your schema from the the list of registries.
1. Choose one or more schemas from the list, by checking the boxes.
1. In the **Action** menu, choose **Delete schema**.
1. Enter the text **Delete** in the field to confirm deletion.
1. Choose **Delete**.
The schema versions you specified are deleted from the registry.
# Deleting a registry
You may want to delete a registry when the schemas it contains should no longer be organized under that registry. You will need to reassign those schemas to another registry.
Deleting one or more registries is a permanent action that cannot be undone. Make sure that the registry or registries no longer needed.
The default registry can be deleted using the AWS CLI.
**AWS Glue API**
To delete the entire registry including the schema and all of its versions, call the [DeleteRegistry action (Python: delete\$1registry)](aws-glue-api-schema-registry-api.md#aws-glue-api-schema-registry-api-DeleteRegistry) API. Specify a `RegistryId` structure to identify the registry.
For example:
```
aws glue delete-registry --registry-id RegistryArn="arn:aws:glue:us-east-2:901234567890:registry/registryName1"
```
```
aws glue delete-registry --registry-id RegistryName="TestRegistry-deletebyname"
```
To get the status of the delete operation, you can call the `GetRegistry` API after the asynchronous call.
**AWS Glue console**
To delete a registry from the AWS Glue console:
1. Sign in to the AWS Management Console and open the AWS Glue console at [https://console.aws.amazon.com/glue/](https://console.aws.amazon.com/glue\).
1. In the navigation pane, under **Data catalog**, choose **Schema registries**.
1. Choose a registry from the list, by checking a box.
1. In the **Action** menu, choose **Delete registry**.
1. Enter the text **Delete** in the field to confirm deletion.
1. Choose **Delete**.
The registries you selected are deleted from AWS Glue.
## IAM examples for serializers
**Note**
AWS managed policies grant necessary permissions for common use cases. For information on using managed policies to manage the schema registry, see [AWS managed (predefined) policies for AWS Glue](security-iam-awsmanpol.md#access-policy-examples-aws-managed).
For serializers, you should create a minimal policy similar to that below to give you the ability to find the `schemaVersionId` for a given schema definition. Note, you should have read permissions on the registry in order to read the schemas in the registry. You can limit the registries that can be read by using the `Resource` clause.
Code example 13:
```
{
"Sid" : "GetSchemaByDefinition",
"Effect" : "Allow",
"Action" :
[
"glue:GetSchemaByDefinition"
],
"Resource" : ["arn:aws:glue:us-east-2:012345678:registry/registryname-1",
"arn:aws:glue:us-east-2:012345678:schema/registryname-1/schemaname-1",
"arn:aws:glue:us-east-2:012345678:schema/registryname-1/schemaname-2"
]
}
```
Further, you can also allow producers to create new schemas and versions by including the following extra methods. Note, you should be able to inspect the registry in order to add/remove/evolve the schemas inside it. You can limit the registries that can be inspected by using the `Resource` clause.
Code example 14:
```
{
"Sid" : "RegisterSchemaWithMetadata",
"Effect" : "Allow",
"Action" :
[
"glue:GetSchemaByDefinition",
"glue:CreateSchema",
"glue:RegisterSchemaVersion",
"glue:PutSchemaVersionMetadata",
],
"Resource" : ["arn:aws:glue:aws-region:123456789012:registry/registryname-1",
"arn:aws:glue:aws-region:123456789012:schema/registryname-1/schemaname-1",
"arn:aws:glue:aws-region:123456789012:schema/registryname-1/schemaname-2"
]
}
```
## IAM examples for deserializers
For deserializers (consumer side), you should create a policy similar to that below to allow the deserializer to fetch the schema from the Schema Registry for deserialization. Note, you should be able to inspect the registry in order to fetch the schemas inside it.
Code example 15:
```
{
"Sid" : "GetSchemaVersion",
"Effect" : "Allow",
"Action" :
[
"glue:GetSchemaVersion"
],
"Resource" : ["*"]
}
```
## Private connectivity using AWS PrivateLink
You can use AWS PrivateLink to connect your data producer’s VPC to AWS Glue by defining an interface VPC endpoint for AWS Glue. When you use a VPC interface endpoint, communication between your VPC and AWS Glue is conducted entirely within the AWS network. For more information, see [Using AWS Glue with VPC Endpoints](https://docs.aws.amazon.com/glue/latest/dg/vpc-endpoint.html).
# Accessing Amazon CloudWatch metrics
Amazon CloudWatch metrics are available as part of CloudWatch’s free tier. You can access these metrics in the CloudWatch console. API-Level metrics include CreateSchema (Success and Latency), GetSchemaByDefinition, (Success and Latency), GetSchemaVersion (Success and Latency), RegisterSchemaVersion (Success and Latency), PutSchemaVersionMetadata (Success and Latency). Resource-level metrics include Registry.ThrottledByLimit, SchemaVersion.ThrottledByLimit, SchemaVersion.Size.
# Sample CloudFormation template for schema registry
The following is a sample template for creating Schema Registry resources in CloudFormation. To create this stack in your account, copy the above template into a file `SampleTemplate.yaml`, and run the following command:
```
aws cloudformation create-stack --stack-name ABCSchemaRegistryStack --template-body "'cat SampleTemplate.yaml'"
```
This example uses `AWS::Glue::Registry` to create a registry, `AWS::Glue::Schema` to create a schema, `AWS::Glue::SchemaVersion` to create a schema version, and `AWS::Glue::SchemaVersionMetadata` to populate schema version metadata.
```
Description: "A sample CloudFormation template for creating Schema Registry resources."
Resources:
ABCRegistry:
Type: "AWS::Glue::Registry"
Properties:
Name: "ABCSchemaRegistry"
Description: "ABC Corp. Schema Registry"
Tags:
Project: "Foo"
ABCSchema:
Type: "AWS::Glue::Schema"
Properties:
Registry:
Arn: !Ref ABCRegistry
Name: "TestSchema"
Compatibility: "NONE"
DataFormat: "AVRO"
SchemaDefinition: >
{"namespace":"foo.avro","type":"record","name":"user","fields":[{"name":"name","type":"string"},{"name":"favorite_number","type":"int"}]}
Tags:
Project: "Foo"
SecondSchemaVersion:
Type: "AWS::Glue::SchemaVersion"
Properties:
Schema:
SchemaArn: !Ref ABCSchema
SchemaDefinition: >
{"namespace":"foo.avro","type":"record","name":"user","fields":[{"name":"status","type":"string", "default":"ON"}, {"name":"name","type":"string"},{"name":"favorite_number","type":"int"}]}
FirstSchemaVersionMetadata:
Type: "AWS::Glue::SchemaVersionMetadata"
Properties:
SchemaVersionId: !GetAtt ABCSchema.InitialSchemaVersionId
Key: "Application"
Value: "Kinesis"
SecondSchemaVersionMetadata:
Type: "AWS::Glue::SchemaVersionMetadata"
Properties:
SchemaVersionId: !Ref SecondSchemaVersion
Key: "Application"
Value: "Kinesis"
```
# Integrating with AWS Glue Schema Registry
These sections describe integrations with AWS Glue schema registry. The examples in these section show a schema with AVRO data format. For more examples, including schemas with JSON data format, see the integration tests and ReadMe information in the [AWS Glue Schema Registry open source repository](https://github.com/awslabs/aws-glue-schema-registry).
**Topics**
+ [Use case: Connecting Schema Registry to Amazon MSK or Apache Kafka](#schema-registry-integrations-amazon-msk)
+ [Use case: Integrating Amazon Kinesis Data Streams with the AWS Glue Schema Registry](#schema-registry-integrations-kds)
+ [Use case: Amazon Managed Service for Apache Flink](#schema-registry-integrations-kinesis-data-analytics-apache-flink)
+ [Use Case: Integration with AWS Lambda](#schema-registry-integrations-aws-lambda)
+ [Use case: AWS Glue Data Catalog](#schema-registry-integrations-aws-glue-data-catalog)
+ [Use case: AWS Glue streaming](#schema-registry-integrations-aws-glue-streaming)
+ [Use case: Apache Kafka Streams](#schema-registry-integrations-apache-kafka-streams)
## Use case: Connecting Schema Registry to Amazon MSK or Apache Kafka
Let's assume you are writing data to an Apache Kafka topic, and you can follow these steps to get started.
1. Create an Amazon Managed Streaming for Apache Kafka (Amazon MSK) or Apache Kafka cluster with at least one topic. If creating an Amazon MSK cluster, you can use the AWS Management Console. Follow these instructions: [Getting Started Using Amazon MSK](https://docs.aws.amazon.com/msk/latest/developerguide/getting-started.html) in the *Amazon Managed Streaming for Apache Kafka Developer Guide*.
1. Follow the [Installing SerDe Libraries](schema-registry-gs-serde.md) step above.
1. To create schema registries, schemas, or schema versions, follow the instructions under the [Getting started with schema registry](schema-registry-gs.md) section of this document.
1. Start your producers and consumers to use the Schema Registry to write and read records to/from the Amazon MSK or Apache Kafka topic. Example producer and consumer code can be found in [the ReadMe file](https://github.com/awslabs/aws-glue-schema-registry/blob/master/README.md) from the Serde libraries. The Schema Registry library on the producer will automatically serialize the record and decorate the record with a schema version ID.
1. If the schema of this record has been inputted, or if auto-registration is turned on, then the schema will have been registered in the Schema Registry.
1. The consumer reading from the Amazon MSK or Apache Kafka topic, using the AWS Glue Schema Registry library, will automatically lookup the schema from the Schema Registry.
## Use case: Integrating Amazon Kinesis Data Streams with the AWS Glue Schema Registry
This integration requires that you have an existing Amazon Kinesis data stream. For more information, see [Getting Started with Amazon Kinesis Data Streams](https://docs.aws.amazon.com/streams/latest/dev/getting-started.html) in the *Amazon Kinesis Data Streams Developer Guide*.
There are two ways that you can interact with data in a Kinesis data stream.
+ Through the Kinesis Producer Library (KPL) and Kinesis Client Library (KCL) libraries in Java. Multi-language support is not provided.
+ Through the `PutRecords`, `PutRecord`, and `GetRecords` Kinesis Data Streams APIs available in the AWS SDK for Java.
If you currently use the KPL/KCL libraries, we recommend continuing to use that method. There are updated KCL and KPL versions with Schema Registry integrated, as shown in the examples. Otherwise, you can use the sample code to leverage the AWS Glue Schema Registry if using the KDS APIs directly.
Schema Registry integration is only available with KPL v0.14.2 or later and with KCL v2.3 or later. Schema Registry integration with JSON data format is available with KPL v0.14.8 or later and with KCL v2.3.6 or later.
### Interacting with Data Using Kinesis SDK V2
This section describes interacting with Kinesis using Kinesis SDK V2
```
// Example JSON Record, you can construct a AVRO record also
private static final JsonDataWithSchema record = JsonDataWithSchema.builder(schemaString, payloadString);
private static final DataFormat dataFormat = DataFormat.JSON;
//Configurations for Schema Registry
GlueSchemaRegistryConfiguration gsrConfig = new GlueSchemaRegistryConfiguration("us-east-1");
GlueSchemaRegistrySerializer glueSchemaRegistrySerializer =
new GlueSchemaRegistrySerializerImpl(awsCredentialsProvider, gsrConfig);
GlueSchemaRegistryDataFormatSerializer dataFormatSerializer =
new GlueSchemaRegistrySerializerFactory().getInstance(dataFormat, gsrConfig);
Schema gsrSchema =
new Schema(dataFormatSerializer.getSchemaDefinition(record), dataFormat.name(), "MySchema");
byte[] serializedBytes = dataFormatSerializer.serialize(record);
byte[] gsrEncodedBytes = glueSchemaRegistrySerializer.encode(streamName, gsrSchema, serializedBytes);
PutRecordRequest putRecordRequest = PutRecordRequest.builder()
.streamName(streamName)
.partitionKey("partitionKey")
.data(SdkBytes.fromByteArray(gsrEncodedBytes))
.build();
shardId = kinesisClient.putRecord(putRecordRequest)
.get()
.shardId();
GlueSchemaRegistryDeserializer glueSchemaRegistryDeserializer = new GlueSchemaRegistryDeserializerImpl(awsCredentialsProvider, gsrConfig);
GlueSchemaRegistryDataFormatDeserializer gsrDataFormatDeserializer =
glueSchemaRegistryDeserializerFactory.getInstance(dataFormat, gsrConfig);
GetShardIteratorRequest getShardIteratorRequest = GetShardIteratorRequest.builder()
.streamName(streamName)
.shardId(shardId)
.shardIteratorType(ShardIteratorType.TRIM_HORIZON)
.build();
String shardIterator = kinesisClient.getShardIterator(getShardIteratorRequest)
.get()
.shardIterator();
GetRecordsRequest getRecordRequest = GetRecordsRequest.builder()
.shardIterator(shardIterator)
.build();
GetRecordsResponse recordsResponse = kinesisClient.getRecords(getRecordRequest)
.get();
List