# Integración con AWS Glue Schema Registry
consumerRecords = new ArrayList<>();
List recordsFromKinesis = recordsResponse.records();
for (int i = 0; i < recordsFromKinesis.size(); i++) {
byte[] consumedBytes = recordsFromKinesis.get(i)
.data()
.asByteArray();
Schema gsrSchema = glueSchemaRegistryDeserializer.getSchema(consumedBytes);
Object decodedRecord = gsrDataFormatDeserializer.deserialize(ByteBuffer.wrap(consumedBytes),
gsrSchema.getSchemaDefinition());
consumerRecords.add(decodedRecord);
}
```
### Interacción con los datos mediante las bibliotecas KPL/KCL
f = producer.addUserRecord(
config.getStreamName(), TIMESTAMP, Utils.randomExplicitHashKey(), data, gsrSchema);
private static ByteBuffer getDataToSend() {
org.apache.avro.Schema avroSchema =
new org.apache.avro.Schema.Parser().parse(SCHEMA_DEFINITION);
GenericRecord user = new GenericData.Record(avroSchema);
user.put("name", "Emily");
user.put("favorite_number", 32);
user.put("favorite_color", "green");
ByteArrayOutputStream outBytes = new ByteArrayOutputStream();
Encoder encoder = EncoderFactory.get().directBinaryEncoder(outBytes, null);
new GenericDatumWriter<>(avroSchema).write(user, encoder);
encoder.flush();
return ByteBuffer.wrap(outBytes.toByteArray());
}
```
#### Configuración de Kinesis Client Library
com.amazonaws
aws-java-sdk-bom
1.11.884
pom
import
com.amazonaws
aws-java-sdk-kinesis
software.amazon.glue
schema-registry-serde
1.1.5
com.fasterxml.jackson.dataformat
jackson-dataformat-cbor
2.11.3
```
1. En el productor, agregue información de encabezado de esquema con la API `PutRecords` o `PutRecord` en Kinesis Data Streams.
```
//The following lines add a Schema Header to the record
com.amazonaws.services.schemaregistry.common.Schema awsSchema =
new com.amazonaws.services.schemaregistry.common.Schema(schemaDefinition, DataFormat.AVRO.name(),
schemaName);
GlueSchemaRegistrySerializerImpl glueSchemaRegistrySerializer =
new GlueSchemaRegistrySerializerImpl(DefaultCredentialsProvider.builder().build(), new GlueSchemaRegistryConfiguration(getConfigs()));
byte[] recordWithSchemaHeader =
glueSchemaRegistrySerializer.encode(streamName, awsSchema, recordAsBytes);
```
1. En el productor, use la API `PutRecords` o `PutRecord` para poner el registro en el flujo de datos.
1. En el consumidor, elimine el registro de esquema del encabezado y serialice un registro de esquemas de Avro.
```
//The following lines remove Schema Header from record
GlueSchemaRegistryDeserializerImpl glueSchemaRegistryDeserializer =
new GlueSchemaRegistryDeserializerImpl(DefaultCredentialsProvider.builder().build(), getConfigs());
byte[] recordWithSchemaHeaderBytes = new byte[recordWithSchemaHeader.remaining()];
recordWithSchemaHeader.get(recordWithSchemaHeaderBytes, 0, recordWithSchemaHeaderBytes.length);
com.amazonaws.services.schemaregistry.common.Schema awsSchema =
glueSchemaRegistryDeserializer.getSchema(recordWithSchemaHeaderBytes);
byte[] record = glueSchemaRegistryDeserializer.getData(recordWithSchemaHeaderBytes);
//The following lines serialize an AVRO schema record
if (DataFormat.AVRO.name().equals(awsSchema.getDataFormat())) {
Schema avroSchema = new org.apache.avro.Schema.Parser().parse(awsSchema.getSchemaDefinition());
Object genericRecord = convertBytesToRecord(avroSchema, record);
System.out.println(genericRecord);
}
```
#### Interacción con datos mediante las API de Kinesis Data Streams
datumReader = new GenericDatumReader<>(avroSchema);
Decoder decoder = DecoderFactory.get().binaryDecoder(record, null);
GenericRecord genericRecord = datumReader.read(null, decoder);
return genericRecord;
}
private Map getMetadata() {
Map metadata = new HashMap<>();
metadata.put("event-source-1", "topic1");
metadata.put("event-source-2", "topic2");
metadata.put("event-source-3", "topic3");
metadata.put("event-source-4", "topic4");
metadata.put("event-source-5", "topic5");
return metadata;
}
private GlueSchemaRegistryConfiguration getConfigs() {
GlueSchemaRegistryConfiguration configs = new GlueSchemaRegistryConfiguration(regionName);
configs.setSchemaName(schemaName);
configs.setAutoRegistration(true);
configs.setMetadata(getMetadata());
return configs;
}
private Object getTestRecord() throws IOException {
GenericRecord genericRecord;
Schema.Parser parser = new Schema.Parser();
Schema avroSchema = parser.parse(new File(AVRO_USER_SCHEMA_FILE));
genericRecord = new GenericData.Record(avroSchema);
genericRecord.put("name", "testName");
genericRecord.put("favorite_number", 99);
genericRecord.put("favorite_color", "red");
return genericRecord;
}
}
```
## Caso de uso: Amazon Managed Service para Apache Flink
software.amazon.glue
schema-registry-flink-serde
1.0.0
```
#### Integración de Kafka o Amazon MSK con Apache Flink
configs = new HashMap<>();
configs.put(AWSSchemaRegistryConstants.AWS_REGION, "aws-region");
configs.put(AWSSchemaRegistryConstants.SCHEMA_AUTO_REGISTRATION_SETTING, true);
configs.put(AWSSchemaRegistryConstants.AVRO_RECORD_TYPE, AvroRecordType.GENERIC_RECORD.getName());
FlinkKafkaConsumer consumer = new FlinkKafkaConsumer<>(
topic,
// block 2
GlueSchemaRegistryAvroDeserializationSchema.forGeneric(schema, configs),
properties);
FlinkKafkaProducer producer = new FlinkKafkaProducer<>(
topic,
// block 3
GlueSchemaRegistryAvroSerializationSchema.forGeneric(schema, topic, configs),
properties);
DataStream stream = env.addSource(consumer);
stream.addSink(producer);
env.execute();
```
#### Integración de Kinesis Data Streams con Apache Flink
configs = new HashMap<>();
configs.put(AWSSchemaRegistryConstants.AWS_REGION, "aws-region");
configs.put(AWSSchemaRegistryConstants.SCHEMA_AUTO_REGISTRATION_SETTING, true);
configs.put(AWSSchemaRegistryConstants.AVRO_RECORD_TYPE, AvroRecordType.GENERIC_RECORD.getName());
FlinkKinesisConsumer consumer = new FlinkKinesisConsumer<>(
streamName,
// block 2
GlueSchemaRegistryAvroDeserializationSchema.forGeneric(schema, configs),
properties);
FlinkKinesisProducer producer = new FlinkKinesisProducer<>(
// block 3
GlueSchemaRegistryAvroSerializationSchema.forGeneric(schema, topic, configs),
properties);
producer.setDefaultStream(streamName);
producer.setDefaultPartition("0");
DataStream stream = env.addSource(consumer);
stream.addSink(producer);
env.execute();
```
## Caso de uso: integración con AWS Lambda
source = builder.stream("avro-input");
```
1. Procese los registros de datos (el ejemplo filtra aquellos registros cuyo valor de color\$1favorito es rosa o cuyo valor de cantidad es 15).
```
final KStream result = source
.filter((key, value) -> !"pink".equals(String.valueOf(value.get("favorite_color"))));
.filter((key, value) -> !"15.0".equals(String.valueOf(value.get("amount"))));
```
1. Escriba los resultados en el tema avro-output.
```
result.to("avro-output");
```
1. Inicie la aplicación Apache Kafka Streams.
```
KafkaStreams streams = new KafkaStreams(builder.build(), props);
streams.start();
```
#### Resultados de implementación
/target/dependency/*"
```
1. Agregue el siguiente comando a la sección **Launch mode (Modo de lanzamiento)** en **kafka-run-class.sh**
```
-cp $CLASSPATH:"/target/dependency/*"
```
Debería tener un aspecto similar al siguiente:
```
# Launch mode
if [ "x$DAEMON_MODE" = "xtrue" ]; then
nohup "$JAVA" $KAFKA_HEAP_OPTS $KAFKA_JVM_PERFORMANCE_OPTS $KAFKA_GC_LOG_OPTS $KAFKA_JMX_OPTS $KAFKA_LOG4J_OPTS -cp $CLASSPATH:"/Users/johndoe/aws-glue-schema-registry/target/dependency/*" $KAFKA_OPTS "$@" > "$CONSOLE_OUTPUT_FILE" 2>&1 < /dev/null &
else
exec "$JAVA" $KAFKA_HEAP_OPTS $KAFKA_JVM_PERFORMANCE_OPTS $KAFKA_GC_LOG_OPTS $KAFKA_JMX_OPTS $KAFKA_LOG4J_OPTS -cp $CLASSPATH:"/Users/johndoe/aws-glue-schema-registry/target/dependency/*" $KAFKA_OPTS "$@"
fi
```
1. Si usa bash, ejecute los siguientes comandos para configurar su CLASSPATH en su bash\$1profile. Para cualquier otro shell, actualice el entorno en consecuencia.
```
echo 'export GSR_LIB_BASE_DIR=<>' >>~/.bash_profile
echo 'export GSR_LIB_VERSION=1.0.0' >>~/.bash_profile
echo 'export KAFKA_HOME=' >>~/.bash_profile
echo 'export CLASSPATH=$CLASSPATH:$GSR_LIB_BASE_DIR/avro-kafkaconnect-converter/target/schema-registry-kafkaconnect-converter-$GSR_LIB_VERSION.jar:$GSR_LIB_BASE_DIR/common/target/schema-registry-common-$GSR_LIB_VERSION.jar:$GSR_LIB_BASE_DIR/avro-serializer-deserializer/target/schema-registry-serde-$GSR_LIB_VERSION.jar' >>~/.bash_profile
source ~/.bash_profile
```
1. (Opcional) si desea realizar una prueba con un origen de archivo simple, clone el conector de origen del archivo.
```
git clone https://github.com/mmolimar/kafka-connect-fs.git
cd kafka-connect-fs/
```
1. En la configuración del conector de origen, edite el formato de datos a Avro, el lector de archivos a `AvroFileReader` y actualice un objeto Avro de ejemplo desde la ruta del archivo de la que está leyendo. Por ejemplo:
```
vim config/kafka-connect-fs.properties
```
```
fs.uris=
policy.regexp=^.*\.avro$
file_reader.class=com.github.mmolimar.kafka.connect.fs.file.reader.AvroFileReader
```
1. Instale el conector de origen.
```
mvn clean package
echo "export CLASSPATH=\$CLASSPATH:\"\$(find target/ -type f -name '*.jar'| grep '\-package' | tr '\n' ':')\"" >>~/.bash_profile
source ~/.bash_profile
```
1. Actualice las propiedades del receptor en `/config/connect-file-sink.properties`, actualice el nombre del tema y el nombre del archivo de salida.
```
file=
topics=
```
1. Inicie el conector de origen (en este ejemplo es un conector de origen de archivo).
```
$KAFKA_HOME/bin/connect-standalone.sh $KAFKA_HOME/config/connect-standalone.properties config/kafka-connect-fs.properties
```
1. Ejecute el conector del receptor (en este ejemplo es un conector receptor de archivo).
```
$KAFKA_HOME/bin/connect-standalone.sh $KAFKA_HOME/config/connect-standalone.properties $KAFKA_HOME/config/connect-file-sink.properties
```
Para ver un ejemplo de uso de Kafka Connect, mire el script run-local-tests.sh en la carpeta integration-tests (pruebas de integración) en el [repositorio de Github para AWS Glue Schema Registry](https://github.com/awslabs/aws-glue-schema-registry/tree/master/integration-tests).