Wednesday, October 28, 2026
HomeBig DataIntroducing Protocol buffers (protobuf) schema assist in Amazon Glue Schema Registry
Big Data

Introducing Protocol buffers (protobuf) schema assist in Amazon Glue Schema Registry

Yalini
By Yalini
0
245
Introducing Protocol buffers (protobuf) schema assist in Amazon Glue Schema Registry

[ad_1]

AWS Glue Schema Registry now helps Protocol buffers (protobuf) schemas along with JSON and Avro schemas. This enables software groups to make use of protobuf schemas to manipulate the evolution of streaming information and centrally management information high quality from information streams to information lake. AWS Glue Schema Registry gives an open-source library that features Apache-licensed serializers and deserializers for protobuf that combine with Java purposes developed for Apache Kafka, Amazon Managed Streaming for Apache Kafka (Amazon MSK), Amazon Kinesis Knowledge Streams, and Kafka Streams. Just like Avro and JSON schemas, Protocol buffers schemas additionally assist compatibility modes, schema sourcing by way of metadata, auto-registration of schemas, and AWS Identification and Entry Administration (IAM) compatibility.

On this put up, we concentrate on Protocol buffers schema assist in AWS Glue Schema Registry and easy methods to use Protocol buffers schemas in stream processing Java purposes that combine with Apache Kafka, Amazon Managed Streaming for Apache Kafka and Amazon Kinesis Knowledge Streams

Introduction to Protocol buffers

Protocol buffers is a language and platform-neutral, extensible mechanism for serializing and deserializing structured information to be used in communications protocols and information storage. A protobuf message format is outlined within the .proto file. Protobuf is beneficial over different information codecs if you want language interoperability, quicker serialization and deserialization, kind security, schema adherence between information producer and client purposes, and decreased coding effort. With protobuf, you need to use generated code from the schema utilizing the protobuf compiler (protoc) to simply write and skim your information to and from information streams utilizing a wide range of languages. You can even use construct instruments plugins comparable to Maven and Gradle to generate code from protobuf schemas as a part of your CI/CD pipelines. ​We use the next schema for code examples on this put up, which defines an worker with a gRPC service definition to seek out an worker by ID:

Worker.proto

syntax = "proto2";
bundle gsr.proto.put up;

import "google/protobuf/wrappers.proto";
import "google/protobuf/period.proto";
import "google/protobuf/timestamp.proto";
import "google/kind/cash.proto";

service EmployeeSearch {
    rpc FindEmployee(EmployeeSearchParams) returns (Worker);
}
message EmployeeSearchParams {
    required int32 id = 1;
}
message Worker {
    required int32 id = 1;
    required string identify = 2;
    required string deal with = 3;
    required google.protobuf.Int32Value employee_age = 4;
    required google.protobuf.Timestamp start_date = 5;
    required google.protobuf.Length total_time_span_in_company = 6;
    required google.protobuf.BoolValue is_certified = 7;
    required Staff workforce = 8;
    required Mission venture = 9;
    required Position position = 10;
    required google.kind.Cash total_award_value = 11;
}
message Staff {
    required string identify = 1;
    required string location = 2;
}
message Mission {
    required string identify = 1;
    required string state = 2;
}
enum Position {
    MANAGER = 0;
    DEVELOPER = 1;
    ARCHITECT = 2;
}

AWS Glue Schema Registry helps each proto2 and proto3 syntax. The previous protobuf schema utilizing model 2 incorporates three message sorts: Worker, Staff, and Mission utilizing scalar, composite, and enumeration information sorts. Every area within the message definitions has a singular quantity, which is used to establish fields within the message binary format, and shouldn’t be modified as soon as your message kind is in use. In a proto2 message, a area may be required, non-obligatory, or repeated; in proto3, the choices are repeated and non-obligatory. The bundle declaration makes certain generated code is namespaced to keep away from any collisions. Along with scalar, composite, and enumeration sorts, AWS Glue Schema Registry additionally helps protobuf schemas with frequent sorts comparable to Cash, PhoneNumber,Timestamp, Length, and nullable sorts comparable to BoolValue and Int32Value. It additionally helps protobuf schemas with gRPC service definitions with compatibility guidelines, comparable to EmployeeSearch, within the previous schema. To study extra in regards to the Protocol buffers, consult with its documentation.

Supported Protocol buffers specification and options

AWS Glue Schema Registry helps all of the options of Protocol buffers for variations 2 and three apart from teams, extensions, and importing definitions. AWS Glue Schema Registry APIs and its open-source library helps the newest protobuf runtime model. The protobuf schema operations in AWS Glue Schema Registry are supported by way of the AWS Administration Console, AWS Command Line Interface (AWS CLI), AWS Glue Schema Registry API, AWS SDK, and AWS CloudFormation.

How AWS Glue Schema Registry works

The next diagram illustrates a high-level view of how AWS Glue Schema Registry works. AWS Glue Schema Registry permits you to register and evolve JSON, Apache Avro, and Protocol buffers schemas with compatibility modes. You possibly can register a number of variations of every schema because the enterprise wants or stream processing software’s necessities evolve. The AWS Glue Schema Registry open-source library gives JSON, Avro, and protobuf serializers and deserializers that you just configure in producer and client stream processing purposes, as proven within the following diagram. The open-source library additionally helps non-obligatory compression and caching configuration to avoid wasting on information transfers.

Introducing Protocol buffers (protobuf) schema assist in Amazon Glue Schema Registry

To accommodate numerous enterprise use instances, AWS Glue Schema Registry helps a number of compatibility modes. For instance, if a client software is up to date to a brand new schema model however continues to be in a position to eat and course of messages primarily based on the earlier model of the identical schema, then the schema is backward-compatible. Nevertheless, if a schema model has bumped up within the producer software and the buyer software just isn’t up to date but however can nonetheless eat and course of the outdated and new message, then the schema is configured as forward-compatible. For extra info, consult with How the Schema Registry Works.

Create a Protocol buffers schema in AWS Glue Schema Registry

On this part, we create a protobuf schema in AWS Glue Schema Registry by way of the console and AWS CLI.

Create a schema by way of the console

Be sure to have the required AWS Glue Schema Registry IAM permissions.

  1. On the AWS Glue console, select Schema registries within the navigation pane.
  2. Click on Add registry.
  3. For Registry identify, enter employee-schema-registry.
  4. Click on Add Registry.
  5. After the registry is created, click on Add schema to register a brand new schema.
  6. For Schema identify, enter Worker.proto.

The schema have to be both Worker.proto or Worker if the protobuf schema doesn’t have the choices choice java_multiple_files = true; and choice java_outer_classname = "<Outer class identify>"; and for those who determine to make use of protobuf schema generated code (POJOs) in your stream processing purposes. We cowl this with an instance in a subsequent part of this put up.­ For extra info on protobuf choices, consult with Choices.

  1. For Registry, select the registry employee-schema-registry.
  2. For Knowledge format, select Protocol buffers.
  3. For Compatibility mode, select Backward.

You possibly can select different compatibility modes as per your use case.

  1. For First schema model, enter the previous protobuf schema, then click on Create schema and model.

After the schema is registered efficiently, its standing will likely be Accessible, as proven within the following screenshot.

Create a schema by way of the AWS CLI

Be sure to have IAM credentials with AWS Glue Schema Registry permissions.

  1. Run the next AWS CLI command to create a schema registry employee-schema-registry (for this put up, we use the Area us-east-2): 
    aws glue create-registry 
--registry-name employee-schema-registry 
--region us-east-2

The AWS CLI command returns the newly created schema registry ARN in response.

  1. Copy the RegistryArn worth from the response to make use of within the following AWS CLI command.
  2. Within the following command, use the previous protobuf schema and schema identify Worker.proto: 
    aws glue create-schema --schema-name Worker.proto 
--registry-id RegistryArn=<Schema Registry ARN that you just copied from response of create registry CLI command> 
--compatibility BACKWARD 
--data-format PROTOBUF 
--schema-definition file:///<project-directory>/Worker.proto 
--region us-east-2

You can even use AWS CloudFormation to create schemas in AWS Glue Schema Registry.

Utilizing a Protocol buffers schema with Amazon MSK and Kinesis Knowledge Streams

Like Apache Avro’s SpecificRecord and GenericRecord, protobuf additionally helps working with POJOs to make sure kind security and DynamicMessage to create generic information producer and client purposes. The next examples showcase the usage of a protobuf schema registered in AWS Glue Schema Registry with Kafka and Kinesis Knowledge Streams producer and client purposes.

Use a protobuf schema with Amazon MSK

Create an Amazon MSK or Apache Kafka cluster with a subject referred to as protobuf-demo-topic. If creating an Amazon MSK cluster, you need to use the console. For directions, consult with Getting Began Utilizing Amazon MSK.

Use protobuf schema-generated POJOs

To make use of protobuf schema-generated POJOs, full the next steps:

  1. Set up the protobuf compiler (protoc) in your native machine from GitHub and add it within the PATH variable.
  2. Add the next plugin configuration to your software’s pom.xml file. We use the xolstice protobuf Maven plugin for this put up to generate code from the protobuf schema. 
    <plugin>
   <!-- https://www.xolstice.org/protobuf-maven-plugin/utilization.html -->
   <groupId>org.xolstice.maven.plugins</groupId>
   <artifactId>protobuf-maven-plugin</artifactId>
   <model>0.6.1</model>
   <configuration>
       <protoSourceRoot>${basedir}/src/major/sources/proto</protoSourceRoot>
       <outputDirectory>${basedir}/src/major/java</outputDirectory>
       <clearOutputDirectory>false</clearOutputDirectory>
   </configuration>
   <executions>
       <execution>
           <targets>
               <purpose>compile</purpose>
           </targets>
       </execution>
   </executions>
</plugin>

  3. Add the next dependencies to your software’s pom.xml file: 
    <!-- https://mvnrepository.com/artifact/com.google.protobuf/protobuf-java -->
<dependency>
   <groupId>com.google.protobuf</groupId>
   <artifactId>protobuf-java</artifactId>
   <model>3.19.4</model>
</dependency>

<!-- https://mvnrepository.com/artifact/software program.amazon.glue/schema-registry-serde -->
<dependency>
   <groupId>software program.amazon.glue</groupId>
   <artifactId>schema-registry-serde</artifactId>
   <model>1.1.9</model>
</dependency>

  4. Create a schema registry employee-schema-registry in AWS Glue Schema Registry and register the Worker.proto protobuf schema with it. Title your schema Worker.proto (or Worker).
  5. Run the next command to generate the code from Worker.proto. Be sure to have the schema file within the ${basedir}/src/major/sources/proto listing or change it as per your software listing construction within the software’s pom.xml <protoSourceRoot> tag worth:

Subsequent, we configure the Kafka producer publishing protobuf messages to the Kafka matter on Amazon MSK.

  1. Configure the Kafka producer properties:
non-public Properties getProducerConfig() {
    Properties props = new Properties();
    props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, this.bootstrapServers);
    props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
    props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, GlueSchemaRegistryKafkaSerializer.class.getName());
    props.put(AWSSchemaRegistryConstants.DATA_FORMAT, DataFormat.PROTOBUF.identify());
    props.put(AWSSchemaRegistryConstants.AWS_REGION,"us-east-2");
    props.put(AWSSchemaRegistryConstants.REGISTRY_NAME, "employee-schema-registry");
    props.put(AWSSchemaRegistryConstants.SCHEMA_NAME, "Worker.proto");
    props.put(AWSSchemaRegistryConstants.PROTOBUF_MESSAGE_TYPE, ProtobufMessageType.POJO.getName());
    return props;
}

The VALUE_SERIALIZER_CLASS_CONFIG configuration specifies the AWS Glue Schema Registry serializer, which serializes the protobuf message.

  1. Use the schema-generated code (POJOs) to create a protobuf message: 
    public EmployeeOuterClass.Worker createEmployeeRecord(int employeeId){
    EmployeeOuterClass.Worker worker =
            EmployeeOuterClass.Worker.newBuilder()
                    .setId(employeeId)
                    .setName("Dummy")
                    .setAddress("Melbourne, Australia")
                    .setEmployeeAge(Int32Value.newBuilder().setValue(32).construct())
                    .setStartDate(Timestamp.newBuilder().setSeconds(235234532434L).construct())
                    .setTotalTimeSpanInCompany(Length.newBuilder().setSeconds(3453245345L).construct())
                    .setIsCertified(BoolValue.newBuilder().setValue(true).construct())
                    .setRole(EmployeeOuterClass.Position.ARCHITECT)
                    .setProject(EmployeeOuterClass.Mission.newBuilder()
                            .setName("Protobuf Schema Demo")
                            .setState("GA").construct())
                    .setTotalAwardValue(Cash.newBuilder()
                                        .setCurrencyCode("USD")
                                        .setUnits(5)
                                        .setNanos(50000).construct())
                    .setTeam(EmployeeOuterClass.Staff.newBuilder()
                            .setName("Options Architects")
                            .setLocation("Australia").construct()).construct();
    return worker;
}

  2. Publish the protobuf messages to the protobuf-demo-topic matter on Amazon MSK: 
    public void startProducer() throws InterruptedException {
    String matter = "protobuf-demo-topic";
    KafkaProducer<String, EmployeeOuterClass.Worker> producer = new KafkaProducer<String, EmployeeOuterClass.Worker>(getProducerConfig());
    logger.information("Beginning to ship data...");
    int employeeId = 0;
    whereas(employeeId < 100)
    {
        EmployeeOuterClass.Worker particular person = createEmployeeRecord(employeeId);
        String key = "key-" + employeeId;
        ProducerRecord<String,  EmployeeOuterClass.Worker> document = new ProducerRecord<String,  EmployeeOuterClass.Worker>(matter, key, particular person);
        producer.ship(document, new ProducerCallback());
        employeeId++;
    }
}
non-public class ProducerCallback implements Callback {
    @Override
    public void onCompletion(RecordMetadata recordMetaData, Exception e){
        if (e == null) {
            logger.information("Acquired new metadata. n" +
                    "Subject:" + recordMetaData.matter() + "n" +
                    "Partition: " + recordMetaData.partition() + "n" +
                    "Offset: " + recordMetaData.offset() + "n" +
                    "Timestamp: " + recordMetaData.timestamp());
        }
        else {
            logger.information("There's been an error from the Producer facet");
            e.printStackTrace();
        }
    }
}

  3. Begin the Kafka producer: 
    public static void major(String args[]) throws InterruptedException {
    ProducerProtobuf producer = new ProducerProtobuf();
    producer.startProducer();
}

  4. Within the Kafka client software’s pom.xml, add the identical plugin and dependencies because the Kafka producer’s pom.xml.

Subsequent, we configure the Kafka client consuming protobuf messages from the Kafka matter on Amazon MSK.

  1. Configure the Kafka client properties: 
    non-public Properties getConsumerConfig() {
    Properties props = new Properties();
    props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, this.bootstrapServers);
    props.put(ConsumerConfig.GROUP_ID_CONFIG, "protobuf-consumer");
    props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG,"earliest");
    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");
    props.put(AWSSchemaRegistryConstants.PROTOBUF_MESSAGE_TYPE, ProtobufMessageType.POJO.getName());
    return props;
}

The VALUE_DESERIALIZER_CLASS_CONFIG config specifies the AWS Glue Schema Registry deserializer that deserializes the protobuf messages.

  1. Eat the protobuf message (as a POJO) from the protobuf-demo-topic matter on Amazon MSK: 
    public void startConsumer() {
    logger.information("beginning client...");
    String matter = "protobuf-demo-topic";
    KafkaConsumer<String, EmployeeOuterClass.Worker> client = new KafkaConsumer<String, EmployeeOuterClass.Worker>(getConsumerConfig());
    client.subscribe(Collections.singletonList(matter));
    whereas (true) {
        last ConsumerRecords<String, EmployeeOuterClass.Worker> data = client.ballot(Length.ofMillis(1000));
        for (last ConsumerRecord<String, EmployeeOuterClass.Worker> document : data)  Title: " + worker.getName() + " 
    }
}

  2. Begin the Kafka client: 
    public static void major(String args[]){
    ConsumerProtobuf client = new ConsumerProtobuf();
    client.startConsumer();
}

Use protobuf’s DynamicMessage

You should utilize DynamicMessage to create generic producer and client purposes with out producing the code from the protobuf schema. To make use of DynamicMessage, you first must create a protobuf schema file descriptor.

  1. Generate a file descriptor from the protobuf schema utilizing the next command: 
    protoc --include_imports --proto_path=proto --descriptor_set_out=proto/Employeeproto.desc proto/Worker.proto

The choice --descritor_set_out has the descriptor file identify that this command generates. The protobuf schema Worker.proto is within the proto listing.

  1. Be sure to have created a schema registry and registered the previous protobuf schema with it.

Now we configure the Kafka producer publishing DynamicMessage to the Kafka matter on Amazon MSK.

  1. Create the Kafka producer configuration. The PROTOBUF_MESSAGE_TYPE configuration is DYNAMIC_MESSAGE as a substitute of POJO. 
    non-public Properties getProducerConfig() {
   Properties props = new Properties();
   props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, this.bootstrapServers);
   props.put(ProducerConfig.ACKS_CONFIG, "-1");
   props.put(ProducerConfig.CLIENT_ID_CONFIG,"protobuf-dynamicmessage-record-producer");
   props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
   props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,GlueSchemaRegistryKafkaSerializer.class.getName());
   props.put(AWSSchemaRegistryConstants.DATA_FORMAT, DataFormat.PROTOBUF.identify());
   props.put(AWSSchemaRegistryConstants.AWS_REGION,"us-east-2");
   props.put(AWSSchemaRegistryConstants.REGISTRY_NAME, "employee-schema-registry");
   props.put(AWSSchemaRegistryConstants.SCHEMA_NAME, "Worker.proto");
   props.put(AWSSchemaRegistryConstants.PROTOBUF_MESSAGE_TYPE, ProtobufMessageType.DYNAMIC_MESSAGE.getName());
   return props;
    }

  2. Create protobuf dynamic messages and publish them to the Kafka matter on Amazon MSK: 
    public void startProducer() throws Exception {
    Descriptor desc = getDescriptor();
    String matter = "protobuf-demo-topic";
    KafkaProducer<String, DynamicMessage> producer = new KafkaProducer<String, DynamicMessage>(getProducerConfig());
    logger.information("Beginning to ship data...");
    int i = 0;
    whereas (i < 100) {
        DynamicMessage dynMessage = DynamicMessage.newBuilder(desc)
                .setField(desc.findFieldByName("id"), 1234)
                .setField(desc.findFieldByName("identify"), "Dummy Title")
                .setField(desc.findFieldByName("deal with"), "Melbourne, Australia")
                .setField(desc.findFieldByName("employee_age"), Int32Value.newBuilder().setValue(32).construct())
                .setField(desc.findFieldByName("start_date"), Timestamp.newBuilder().setSeconds(235234532434L).construct())
                .setField(desc.findFieldByName("total_time_span_in_company"), Length.newBuilder().setSeconds(3453245345L).construct())
                .setField(desc.findFieldByName("is_certified"), BoolValue.newBuilder().setValue(true).construct())
		.setField(desc.findFieldByName("total_award_value"), Cash.newBuilder().setCurrencyCode("USD")
						.setUnits(1).setNanos(50000).construct())
                .setField(desc.findFieldByName("workforce"), createTeam(desc.findFieldByName("workforce").getMessageType()))
                .setField(desc.findFieldByName("venture"), createProject(desc.findFieldByName("venture").getMessageType()))
                .setField(desc.findFieldByName("position"), desc.findFieldByName("position").getEnumType().findValueByName("ARCHITECT"))
                .construct();
        String key = "key-" + i;
        ProducerRecord<String, DynamicMessage> document = new ProducerRecord<String, DynamicMessage>(matter, key, dynMessage);
        producer.ship(document, new ProtobufProducer.ProducerCallback());
        Thread.sleep(1000);
        i++;
    }
}
non-public static DynamicMessage createTeam(Descriptor desc) {
    DynamicMessage dynMessage = DynamicMessage.newBuilder(desc)
            .setField(desc.findFieldByName("identify"), "Options Architects")
            .setField(desc.findFieldByName("location"), "Australia")
            .construct();
    return dynMessage;
}

non-public static DynamicMessage createProject(Descriptor desc) {
    DynamicMessage dynMessage = DynamicMessage.newBuilder(desc)
            .setField(desc.findFieldByName("identify"), "Protobuf Schema Demo")
            .setField(desc.findFieldByName("state"), "GA")
            .construct();
    return dynMessage;
}

non-public class ProducerCallback implements Callback {
    @Override
    public void onCompletion(RecordMetadata recordMetaData, Exception e) {
        if (e == null) {
            logger.information("Acquired new metadata. n" +
                    "Subject:" + recordMetaData.matter() + "n" +
                    "Partition: " + recordMetaData.partition() + "n" +
                    "Offset: " + recordMetaData.offset() + "n" +
                    "Timestamp: " + recordMetaData.timestamp());
        } else {
            logger.information("There's been an error from the Producer facet");
            e.printStackTrace();
        }
    }
}

  3. Create a descriptor utilizing the Employeeproto.desc file that we generated from the Worker.proto schema file within the earlier steps: 
    non-public Descriptor getDescriptor() throws Exception {
    InputStream inStream = ProtobufProducer.class.getClassLoader().getResourceAsStream("proto/Employeeproto.desc");
    DescriptorProtos.FileDescriptorSet fileDescSet = DescriptorProtos.FileDescriptorSet.parseFrom(inStream);
    Map<String, DescriptorProtos.FileDescriptorProto> fileDescProtosMap = new HashMap<String, DescriptorProtos.FileDescriptorProto>();
    Checklist<DescriptorProtos.FileDescriptorProto> fileDescProtos = fileDescSet.getFileList();
    for (DescriptorProtos.FileDescriptorProto fileDescProto : fileDescProtos) {
        fileDescProtosMap.put(fileDescProto.getName(), fileDescProto);
    }
    DescriptorProtos.FileDescriptorProto fileDescProto = fileDescProtosMap.get("Worker.proto");
    FileDescriptor[] dependencies = getProtoDependencies(fileDescProtosMap, fileDescProto);
    FileDescriptor fileDesc = FileDescriptor.buildFrom(fileDescProto, dependencies);
    Descriptor desc = fileDesc.findMessageTypeByName("Worker");
    return desc;
}

public static FileDescriptor[] getProtoDependencies(Map<String, FileDescriptorProto> fileDescProtos, 
				  FileDescriptorProto fileDescProto) throws Exception {

    if (fileDescProto.getDependencyCount() == 0)
        return new FileDescriptor[0];

    ProtocolStringList dependencyList = fileDescProto.getDependencyList();
    String[] dependencyArray = dependencyList.toArray(new String[0]);
    int noOfDependencies = dependencyList.dimension();

    FileDescriptor[] dependencies = new FileDescriptor[noOfDependencies];
    for (int i = 0; i < noOfDependencies; i++) {
        FileDescriptorProto dependencyFileDescProto = fileDescProtos.get(dependencyArray[i]);
        FileDescriptor dependencyFileDesc = FileDescriptor.buildFrom(dependencyFileDescProto, 
					     getProtoDependencies(fileDescProtos, dependencyFileDescProto));
        dependencies[i] = dependencyFileDesc;
    }
    return dependencies;
}

  4. Begin the Kafka producer: 
    public static void major(String args[]) throws InterruptedException {
	 ProducerProtobuf producer = new ProducerProtobuf();
         producer.startProducer();
}

Now we configure the Kafka client consuming dynamic messages from the Kaka matter on Amazon MSK.

  1. Enter the next Kafka client configuration: 
    non-public Properties getConsumerConfig() {
    Properties props = new Properties();
    props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, this.bootstrapServers);
    props.put(ConsumerConfig.GROUP_ID_CONFIG, "protobuf-record-consumer");
    props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG,"earliest");
    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");
    props.put(AWSSchemaRegistryConstants.PROTOBUF_MESSAGE_TYPE, ProtobufMessageType.DYNAMIC_MESSAGE.getName());
    return props;
}

  2. Eat protobuf dynamic messages from the Kafka matter protobuf-demo-topic. As a result of we’re utilizing DYNAMIC_MESSAGE, the retrieved objects are of kind DynamicMessage. 
    public void startConsumer() {
    logger.information("beginning client...");
    String matter = "protobuf-demo-topic";
    KafkaConsumer<String, DynamicMessage> client = new KafkaConsumer<String, DynamicMessage>(getConsumerConfig());
    client.subscribe(Collections.singletonList(matter));
    whereas (true) {
        last ConsumerRecords<String, DynamicMessage> data = client.ballot(Length.ofMillis(1000));
        for (last ConsumerRecord<String, DynamicMessage> document : data) {
            for (Descriptors.FieldDescriptor area : document.worth().getAllFields().keySet()) {
                logger.information(area.getName() + ": " + document.worth().getField(area));
            }
        }
    }
}

  3. Begin the Kafka client: 
    public static void major(String args[]){
        ConsumerProtobuf client = new ConsumerProtobuf();
        client.startConsumer();
     }

Use a protobuf schema with Kinesis Knowledge Streams

You should utilize the protobuf schema-generated POJOs with the Kinesis Producer Library (KPL) and Kinesis Shopper Library (KCL).

  1. Set up the protobuf compiler (protoc) in your native machine from GitHub and add it within the PATH variable.
  2. Add the next plugin configuration to your software’s pom.xml file. We’re utilizing the xolstice protobuf Maven plugin for this put up to generate code from the protobuf schema. 
    <plugin>
   <!-- https://www.xolstice.org/protobuf-maven-plugin/utilization.html -->
   <groupId>org.xolstice.maven.plugins</groupId>
   <artifactId>protobuf-maven-plugin</artifactId>
   <model>0.6.1</model>
   <configuration>
       <protoSourceRoot>${basedir}/src/major/sources/proto</protoSourceRoot>
       <outputDirectory>${basedir}/src/major/java</outputDirectory>
       <clearOutputDirectory>false</clearOutputDirectory>
   </configuration>
   <executions>
       <execution>
           <targets>
               <purpose>compile</purpose>
           </targets>
       </execution>
   </executions>
</plugin>

  3. As a result of the KPL and KCL newest variations have the AWS Glue Schema Registry open-source library (schema-registry-serde) and protobuf runtime (protobuf-java) included, you solely want so as to add the next dependencies to your software’s pom.xml: 
    <!-- https://mvnrepository.com/artifact/com.amazonaws/amazon-kinesis-producer -->
<dependency>
    <groupId>com.amazonaws</groupId>
    <artifactId>amazon-kinesis-producer</artifactId>
    <model>0.14.11</model>
	</dependency>
	<!-- https://mvnrepository.com/artifact/software program.amazon.kinesis/amazon-kinesis-client -->
<dependency>
    <groupId>software program.amazon.kinesis</groupId>
    <artifactId>amazon-kinesis-client</artifactId>
    <model>2.4.0version>
</dependency>

  4. Create a schema registry employee-schema-registry and register the Worker.proto protobuf schema with it. Title your schema Worker.proto (or Worker).
  5. Run the next command to generate the code from Worker.proto. Be sure to have the schema file within the ${basedir}/src/major/sources/proto listing or change it as per your software listing construction within the software’s pom.xml <protoSourceRoot> tag worth.

The next Kinesis producer code with the KPL makes use of the Schema Registry open-source library to publish protobuf messages to Kinesis Knowledge Streams.

  1. Begin the Kinesis Knowledge Streams producer: 
    non-public static last String PROTO_SCHEMA_FILE = "proto/Worker.proto";
non-public static last String SCHEMA_NAME = "Worker.proto";
non-public static String REGION_NAME = "us-east-2";
non-public static String REGISTRY_NAME = "employee-schema-registry";
non-public static String STREAM_NAME = "employee_data_stream";
non-public static int NUM_OF_RECORDS = 100;
non-public static String REGISTRY_ENDPOINT = "https://glue.us-east-2.amazonaws.com";

public static void major(String[] args) throws Exception {
     ProtobufKPLProducer producer = new ProtobufKPLProducer();
     producer.startProducer();
 }
}

  2. Configure the Kinesis producer:
public void startProducer() throws Exception {
    logger.information("Beginning KPL shopper with Glue Schema Registry Integration...");
    GlueSchemaRegistryConfiguration schemaRegistryConfig = new GlueSchemaRegistryConfiguration(REGION_NAME);
    schemaRegistryConfig.setCompressionType(AWSSchemaRegistryConstants.COMPRESSION.ZLIB);
    schemaRegistryConfig.setSchemaAutoRegistrationEnabled(false);
    schemaRegistryConfig.setCompatibilitySetting(Compatibility.BACKWARD);
    schemaRegistryConfig.setEndPoint(REGISTRY_ENDPOINT);
    schemaRegistryConfig.setProtobufMessageType(ProtobufMessageType.POJO);
    schemaRegistryConfig.setRegistryName(REGISTRY_NAME);
	
    //Setting Glue Schema Registry configuration in Kinesis Producer Configuration together with different configs
    KinesisProducerConfiguration config = new KinesisProducerConfiguration()
                                        .setRecordMaxBufferedTime(3000)
                                        .setMaxConnections(1)
                                        .setRequestTimeout(60000)
                                        .setRegion(REGION_NAME)
                                        .setRecordTtl(60000)
                                        .setGlueSchemaRegistryConfiguration(schemaRegistryConfig);

    FutureCallback<UserRecordResult> myCallback = new FutureCallback<UserRecordResult>() {
        @Override public void onFailure(Throwable t) {
              t.printStackTrace();
        };
        @Override public void onSuccess(UserRecordResult consequence)  Shard Id : " + consequence.getShardId());
        ;
    };
    
	//Creating schema definition object from the Worker.proto schema file.
    Schema gsrSchema = getSchemaDefinition();
    last KinesisProducer producer = new KinesisProducer(config);
    int employeeCount = 1;
    whereas(true) {
        //Creating and serializing schema generated POJO object (protobuf message)

        EmployeeOuterClass.Worker worker = createEmployeeRecord(employeeCount);
        byte[] serializedBytes = worker.toByteArray();
        ByteBuffer information = ByteBuffer.wrap(serializedBytes);
        Immediate timestamp = Immediate.now();

        //Publishing protobuf message to the Kinesis Knowledge Stream
        ListenableFuture<UserRecordResult> f =
                    producer.addUserRecord(STREAM_NAME,
                                        Lengthy.toString(timestamp.toEpochMilli()),
                                        new BigInteger(128, new Random()).toString(10),
                                        information,
                                        gsrSchema);
        Futures.addCallback(f, myCallback, MoreExecutors.directExecutor());
        employeeCount++;
        if(employeeCount > NUM_OF_RECORDS)
            break;
    }
    Checklist<Future<UserRecordResult>> putFutures = new LinkedList<>();
    for (Future<UserRecordResult> future : putFutures) {
        UserRecordResult userRecordResult = future.get();
        logger.information(userRecordResult.getShardId() + userRecordResult.getSequenceNumber());
    }
}

  1. Create a protobuf message utilizing schema-generated code (POJOs): 
    public EmployeeOuterClass.Worker createEmployeeRecord(int depend){
    EmployeeOuterClass.Worker worker =
            EmployeeOuterClass.Worker.newBuilder()
            .setId(depend)
            .setName("Dummy")
            .setAddress("Melbourne, Australia")
            .setEmployeeAge(Int32Value.newBuilder().setValue(32).construct())
            .setStartDate(Timestamp.newBuilder().setSeconds(235234532434L).construct())
            .setTotalTimeSpanInCompany(Length.newBuilder().setSeconds(3453245345L).construct())
            .setIsCertified(BoolValue.newBuilder().setValue(true).construct())
            .setRole(EmployeeOuterClass.Position.ARCHITECT)
            .setProject(EmployeeOuterClass.Mission.newBuilder()
                        .setName("Protobuf Schema Demo")
                        .setState("GA").construct())
            .setTotalAwardValue(Cash.newBuilder()
                        .setCurrencyCode("USD")
                        .setUnits(5)
                        .setNanos(50000).construct())
            .setTeam(EmployeeOuterClass.Staff.newBuilder()
                        .setName("Options Architects")
                        .setLocation("Australia").construct()).construct();
    return worker;
}

  2. Create the schema definition from Worker.proto: 
    non-public Schema getSchemaDefinition() throws IOException {
    InputStream inputStream = ProtobufKPLProducer.class.getClassLoader().getResourceAsStream(PROTO_SCHEMA_FILE);
    StringBuilder resultStringBuilder = new StringBuilder();
    strive (BufferedReader br = new BufferedReader(new InputStreamReader(inputStream))) {
        String line;
        whereas ((line = br.readLine()) != null) {
            resultStringBuilder.append(line).append("n");
        }
    }
    String schemaDefinition = resultStringBuilder.toString();
    logger.information("Schema Definition " + schemaDefinition);
    Schema gsrSchema =
            new Schema(schemaDefinition, DataFormat.PROTOBUF.toString(), SCHEMA_NAME);
    return gsrSchema;
}

The next is the Kinesis client code with the KCL utilizing the Schema Registry open-source library to eat protobuf messages from the Kinesis Knowledge Streams.

  1. Initialize the applying: 
    public void run(){
    logger.information("Beginning KCL shopper with Glue Schema Registry Integration...");
    Area area = Area.of(ObjectUtils.firstNonNull(REGION_NAME, "us-east-2"));
    KinesisAsyncClient kinesisClient = KinesisClientUtil.createKinesisAsyncClient(KinesisAsyncClient.builder().area(area));
    DynamoDbAsyncClient dynamoClient = DynamoDbAsyncClient.builder().area(area).construct();
    CloudWatchAsyncClient cloudWatchClient = CloudWatchAsyncClient.builder().area(area).construct();

    EmployeeRecordProcessorFactory employeeRecordProcessorFactory = new EmployeeRecordProcessorFactory();
    ConfigsBuilder configsBuilder =
            new ConfigsBuilder(STREAM_NAME,
                    APPLICATION_NAME,
                    kinesisClient,
                    dynamoClient,
                    cloudWatchClient,
                    APPLICATION_NAME,
                    employeeRecordProcessorFactory);

    //Creating Glue Schema Registry configuration and Glue Schema Registry Deserializer object.
    GlueSchemaRegistryConfiguration gsrConfig = new GlueSchemaRegistryConfiguration(area.toString());
    gsrConfig.setEndPoint(REGISTRY_ENDPOINT);
    gsrConfig.setProtobufMessageType(ProtobufMessageType.POJO);
    GlueSchemaRegistryDeserializer glueSchemaRegistryDeserializer =
            new GlueSchemaRegistryDeserializerImpl(DefaultCredentialsProvider.builder().construct(), gsrConfig);
    /*
     Setting Glue Schema Registry deserializer within the Retrieval Config for
     Kinesis Shopper Library to make use of it whereas deserializing the protobuf messages.
     */
    RetrievalConfig retrievalConfig = configsBuilder.retrievalConfig().retrievalSpecificConfig(new PollingConfig(STREAM_NAME, kinesisClient));
    retrievalConfig.glueSchemaRegistryDeserializer(glueSchemaRegistryDeserializer);

    Scheduler scheduler = new Scheduler(
            		configsBuilder.checkpointConfig(),
            		configsBuilder.coordinatorConfig(),
           		configsBuilder.leaseManagementConfig(),
            		configsBuilder.lifecycleConfig(),
            		configsBuilder.metricsConfig(),
            		configsBuilder.processorConfig(),
            		retrievalConfig);

    Thread schedulerThread = new Thread(scheduler);
    schedulerThread.setDaemon(true);
    schedulerThread.begin();

    logger.information("Press enter to shutdown");
    BufferedReader reader = new BufferedReader(new InputStreamReader(System.in));
    strive {
        reader.readLine();
        Future<Boolean> gracefulShutdownFuture = scheduler.startGracefulShutdown();
        logger.information("Ready as much as 20 seconds for shutdown to finish.");
        gracefulShutdownFuture.get(20, TimeUnit.SECONDS);
    } catch (Exception e) {
        logger.information("Interrupted whereas ready for sleek shutdown. Persevering with.");
    }
    logger.information("Accomplished, shutting down now.");
}

  2. Eat protobuf messages from Kinesis Knowledge Streams: 
    public static class EmployeeRecordProcessorFactory implements ShardRecordProcessorFactory {
    @Override
    public ShardRecordProcessor shardRecordProcessor() {
        return new EmployeeRecordProcessor();
    }
}
public static class EmployeeRecordProcessor implements ShardRecordProcessor {
    non-public static last Logger logger = Logger.getLogger(EmployeeRecordProcessor.class.getSimpleName());
    public void initialize(InitializationInput initializationInput) {}
    public void processRecords(ProcessRecordsInput processRecordsInput) {
        strive {
            logger.information("Processing " + processRecordsInput.data().dimension() + " document(s)");
            for (KinesisClientRecord r : processRecordsInput.data())  Position: " + worker.getRole().identify() + " 
        } catch (Exception e) {
            logger.information("Failed whereas processing data. Aborting" + e);
            Runtime.getRuntime().halt(1);
        }
    }
    public void leaseLost(LeaseLostInput leaseLostInput) {. . .}
    public void shardEnded(ShardEndedInput shardEndedInput) {. . .}
    public void shutdownRequested(ShutdownRequestedInput shutdownRequestedInput) {. . .}
}

  3. Begin the Kinesis Knowledge Streams client: 
    non-public static last Logger logger = Logger.getLogger(ProtobufKCLConsumer.class.getSimpleName());
non-public static String REGION_NAME = "us-east-2";
non-public static String STREAM_NAME = "employee_data_stream";
non-public static last String APPLICATION_NAME =  "protobuf-demo-kinesis-kpl-consumer";
non-public static String REGISTRY_ENDPOINT = "https://glue.us-east-2.amazonaws.com";

public static void major(String[] args) throws ParseException {
    new ProtobufKCLConsumer().run();
}

Improve your protobuf schema

We lined examples of knowledge producer and client purposes integrating with Amazon MSK, Apache Kafka, and Kinesis Knowledge Streams, and utilizing a Protocol buffers schema registered with AWS Glue Schema Registry. You possibly can additional improve these examples with schema evolution utilizing the next guidelines, that are supported by AWS Glue Schema Registry. For instance, the next protobuf schema proven is a backward-compatible up to date model of Worker.proto. We’ve got added one other gRPC service definition CreateEmployee beneath EmployeeSearch and added an Non-obligatory area within the Worker message kind. In case you improve the buyer software with this model of the protobuf schema, the buyer software can nonetheless eat outdated and new protobuf messages.

Worker.proto (version-2) 

syntax = "proto2";
bundle gsr.proto.put up;

import "google/protobuf/wrappers.proto";
import "google/protobuf/period.proto";
import "google/protobuf/timestamp.proto";
import "google/protobuf/empty.proto";
import "google/kind/cash.proto";

service EmployeeSearch {
    rpc FindEmployee(EmployeeSearchParams) returns (Worker);
    rpc CreateEmployee(EmployeeSearchParams) returns (google.protobuf.Empty);
}
message EmployeeSearchParams {
    required int32 id = 1;
}
message Worker {
    required int32 id = 1;
    required string identify = 2;
    required string deal with = 3;
    required google.protobuf.Int32Value employee_age = 4;
    required google.protobuf.Timestamp start_date = 5;
    required google.protobuf.Length total_time_span_in_company = 6;
    required google.protobuf.BoolValue is_certified = 7;
    required Staff workforce = 8;
    required Mission venture = 9;
    required Position position = 10;
    required google.kind.Cash total_award_value = 11;
    non-obligatory string title = 12;
}
message Staff {
    required string identify = 1;
    required string location = 2;
}
message Mission {
    required string identify = 1;
    required string state = 2;
}
enum Position {
    MANAGER = 0;
    DEVELOPER = 1;
    ARCHITECT = 2;
}

Conclusion

On this put up, we launched Protocol buffers schema assist in AWS Glue Schema Registry. AWS Glue Schema Registry now helps Apache Avro, JSON, and Protocol buffers schemas with completely different suitable modes. The examples on this put up demonstrated easy methods to use Protocol buffers schemas registered with AWS Glue Schema Registry in stream processing purposes built-in with Apache Kafka, Amazon MSK, and Kinesis Knowledge Streams. We used the schema-generated POJOs for kind security and protobuf’s DynamicMessage to create generic producer and client purposes. The examples on this put up comprise the fundamental elements of the stream processing sample; you possibly can adapt these examples to your use case wants.

To study extra, consult with the next sources:

In regards to the Creator

Vikas Bajaj is a Principal Options Architect at AWS. Vikas works with digital native clients and advises them on expertise structure and options to satisfy strategic enterprise aims.

[ad_2]

Previous article
Enhance distant storage efficiency with Azure Ebsv5 VMs—now usually accessible | Azure Weblog and Updates
Next article
Google Native Provides In-Retailer Merchandise To Assist Docs On How To Get Higher Visibility
Yalini
Yalinihttps://kopivy.com
RELATED ARTICLES

LEAVE A REPLY

Please enter your comment!
Please enter your name here

Most Popular

Load more

Recent Comments

ABOUT US

kopivy is your news, Technology , Software Development, Artificial Intelligence, SEO website. We provide you with the latest breaking news and videos straight from the Tech industry.

POPULAR POSTS

POPULAR CATEGORY

Copyright © 2021 kopivy.com All rights reserved