# AWS Glue Schema Registry **nota** El Registro de esquemas de AWS Glue no es compatible en la siguiente región en la consola de AWS Glue: Oriente Medio (EAU). Con AWS Glue Schema Registry, puede descubrir, controlar y evolucionar de forma centralizada los esquemas de flujo de datos. Un *esquema* define la estructura y el formato de un registro de datos. Con AWS Glue Schema Registry, puede administrar y aplicar esquemas en sus aplicaciones de streaming de datos mediante integraciones convenientes con 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 para Apache Flink](https://aws.amazon.com/kinesis/data-analytics/) y [AWS Lambda](https://aws.amazon.com/lambda/). Schema Registry admite el formato de datos AVRO (v1.11.4), el formato de datos JSON con [formato de esquemas JSON](https://json-schema.org/) para el esquema (especificaciones Draft-04, Draft-06 y Draft-07) con la validación de esquemas JSON mediante el método [biblioteca de Everit](https://github.com/everit-org/json-schema), y admite los búferes de protocolo (Protobuf) en las versiones proto2 y proto3 sin soporte para `extensions` o `groups`, y el lenguaje Java, con otros formatos de datos e idiomas que estarán disponibles en el futuro. Las características admitidas incluyen compatibilidad, fuentes de esquemas mediante metadatos, registro automático de esquemas, compatibilidad con IAM y compresión ZLIB opcional para reducir el almacenamiento y la transferencia de datos. Schema Registry funciona sin servidor y es gratuito. El uso de un esquema como contrato de formato de datos entre productores y consumidores conduce a una mejor gobernanza de los datos, datos de mayor calidad y permite que los consumidores de datos sean resistentes a los cambios posteriores compatibles. Schema Registry permite que sistemas dispares compartan un esquema para la serialización y la deserialización. Por ejemplo, suponga que tiene un productor y un consumidor de datos. El productor conoce el esquema cuando publica los datos. Schema Registry proporciona un serializador y deserializador para determinados sistemas como Amazon MSK o Apache Kafka. Para obtener más información, consulte [Cómo funciona Schema Registry](schema-registry-works.md). **Topics** + [Schemas](#schema-registry-schemas) + [Registries](#schema-registry-registries) + [Compatibilidad y control de versiones de esquemas](#schema-registry-compatibility) + [Bibliotecas Serde de código abierto](#schema-registry-serde-libraries) + [Cuotas del Schema Registry](#schema-registry-quotas) + [Cómo funciona Schema Registry](schema-registry-works.md) + [Introducción a Schema Registry](schema-registry-gs.md) ## Schemas Un *esquema* define la estructura y el formato de un registro de datos. Un esquema es una especificación versionada para publicación, consumo o almacenamiento de confianza de datos. En este esquema de ejemplo para Avro, el formato y la estructura están definidos por el diseño y los nombres de campos, y el formato de los nombres de campos está definido por los tipos de datos (por ejemplo, `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" } ] } } ] } ``` En este ejemplo de esquema JSON Draft-07 para JSON, el formato está definido por la [organización del esquema JSON](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 } } } ``` En este ejemplo de Protobuf, el formato se define mediante la[versión 2 del lenguaje de búferes de protocolo (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 Un *registro* es un contenedor lógico de esquemas. Los registros le permiten organizar sus esquemas, así como administrar el control de acceso para sus aplicaciones. Un registro tiene un nombre de recurso de Amazon (ARN) que le permite organizar y establecer diferentes permisos de acceso a las operaciones de esquema dentro del registro. Puede utilizar el registro predeterminado o crear tantos registros nuevos como sea necesario. **AWS GlueJerarquía de Schema Registry** | | | --- | | [\[See the AWS documentation website for more details\]](http://docs.aws.amazon.com/es_es/glue/latest/dg/schema-registry.html) | | [\[See the AWS documentation website for more details\]](http://docs.aws.amazon.com/es_es/glue/latest/dg/schema-registry.html) | | [\[See the AWS documentation website for more details\]](http://docs.aws.amazon.com/es_es/glue/latest/dg/schema-registry.html) | | [\[See the AWS documentation website for more details\]](http://docs.aws.amazon.com/es_es/glue/latest/dg/schema-registry.html) | ## Compatibilidad y control de versiones de esquemas Cada esquema puede tener varias versiones. El control de versiones se rige por una regla de compatibilidad que se aplica a un esquema. Schema Registry contrasta las solicitudes para registrar nuevas versiones de esquema con esta regla antes de que puedan tener éxito. Una versión de esquema marcada como punto de verificación se utiliza para determinar la compatibilidad de registrar nuevas versiones de un esquema. Cuando se crea un esquema por primera vez, el punto de verificación predeterminado será la primera versión. A medida que el esquema evoluciona con más versiones, puede utilizar CLI/SDK para cambiar el punto de control a una versión de un esquema mediante la herramienta API `UpdateSchema` que cumple con un conjunto de restricciones. En la consola, la edición de la definición de esquema o el modo de compatibilidad cambiará el punto de control a la última versión de forma predeterminada. Los modos de compatibilidad permiten controlar cómo los esquemas pueden o no evolucionar con el tiempo. Estos modos forman el contrato entre las aplicaciones que producen y consumen datos. Cuando se envía una nueva versión de un esquema al registro, la regla de compatibilidad aplicada al nombre del esquema se utiliza para determinar si se puede aceptar la nueva versión. Existen ocho modos de compatibilidad: NONE (NINGUNO), DISABLED (DESACTIVADO), BACKWARD (HACIA ATRÁS), BACKWARD\$1ALL (HACIA ATRÁS\$1TODO), FORWARD (HACIA ADELANTE), FORWARD\$1ALL (HACIA ADELANTE\$1TODO), FULL (COMPLETO), FULL\$1ALL (COMPLETO\$1TODO). En el formato de datos Avro, los campos pueden ser opcionales u obligatorios. Un campo opcional es aquel en el que el `Type` incluye nulo. Los campos obligatorios no tienen nulo como `Type`. En el formato de datos Protobuf, los campos pueden ser opcionales (incluso repetidos) u obligatorios en la sintaxis proto2, mientras que todos los campos son opcionales (incluidos los repetidos) en la sintaxis proto3. Todas las reglas de compatibilidad se determinan en función de la comprensión de las especificaciones de los búferes de protocolo, así como en función de las pautas en la [documentación sobre búferes de protocolo de Google](https://developers.google.com/protocol-buffers/docs/overview#updating). + *NONE (NINGUNO)*: no se aplica ningún modo de compatibilidad. Puede utilizar esta opción en escenarios de desarrollo o si no conoce los modos de compatibilidad que desea aplicar a los esquemas. Cualquier nueva versión que se agregue será aceptada sin someterse a una comprobación de compatibilidad. + *DISABLED (DESACTIVADO)*: esta opción de compatibilidad impide el control de versiones de un esquema en particular. No se pueden agregar nuevas versiones. + *BACKWARD (HACIA ATRÁS)*: se recomienda esta opción de compatibilidad ya que permite a los consumidores leer tanto la versión actual como la anterior del esquema. Puede utilizar esta opción para comprobar la compatibilidad con la última versión del esquema, al eliminar campos o agregar campos adicionales. Un caso de uso típico para HACIA ATRÁS es cuando la aplicación se ha creado para el esquema más reciente. **AVRO** Por ejemplo, suponga que tiene un esquema definido por nombre (obligatorio), apellido (obligatorio), email (obligatorio) y número de teléfono (opcional). Si la siguiente versión del esquema elimina el campo de email obligatorio, esto se registraría correctamente. La compatibilidad HACIA ATRÁS requiere que los consumidores puedan leer la versión actual y anterior del esquema. Sus consumidores podrán leer el nuevo esquema a medida que se ignora el campo de email adicional de los mensajes antiguos. Si tiene una nueva versión de esquema propuesta que agrega un campo obligatorio, por ejemplo, código postal, esto no se registrará correctamente con la compatibilidad HACIA ATRÁS. Los consumidores de la nueva versión no podrían leer mensajes antiguos antes de cambiar el esquema, ya que les faltará el campo de código postal obligatorio. Sin embargo, si el campo de código postal se estableció como opcional en el nuevo esquema, entonces la versión propuesta se registraría correctamente, ya que los consumidores pueden leer el esquema anterior sin el campo de código postal opcional. **JSON** Por ejemplo, supongamos que tiene una versión de esquema definida por nombre (opcional), apellido (opcional), email (opcional) y número de teléfono (opcional). Si la siguiente versión de esquema agrega la propiedad opcional de número de teléfono, esto se registraría correctamente siempre y cuando la versión original del esquema no permita ninguna propiedad adicional al configurar el campo `additionalProperties` en falso. La compatibilidad HACIA ATRÁS requiere que los consumidores puedan leer la versión actual y anterior del esquema. Sus consumidores podrán leer los datos producidos con el esquema original en el que la propiedad del número de teléfono no existe. Si tiene una nueva versión de esquema propuesta que agrega la propiedad de número de teléfono opcional, esto no se registrará correctamente con la compatibilidad HACIA ATRÁS, cuando la versión del esquema original establezca el campo `additionalProperties` a verdadero, es decir, cuando permita cualquier propiedad adicional. Los consumidores de la nueva versión no podrían leer mensajes antiguos antes de cambiar el esquema, ya que no pueden leer datos con propiedad de número de teléfono en un tipo diferente, por ejemplo, cadena en lugar de número. **PROTOBUF** Por ejemplo, suponga que tiene una versión de esquema definida por un mensaje `Person` con `first name` (obligatorio), `last name` (obligatorio), `email` (obligatorio) y `phone number` (opcional) en la sintaxis proto2. En forma semejante a las situaciones de AVRO, si la siguiente versión del esquema elimina el campo de `email` obligatorio, esto se registraría correctamente. La compatibilidad HACIA ATRÁS requiere que los consumidores puedan leer la versión actual y anterior del esquema. Sus consumidores podrán leer el nuevo esquema a medida que se ignora el campo de `email` adicional de los mensajes antiguos. Si tiene una nueva versión de esquema propuesta que agrega un campo obligatorio, por ejemplo, `zip code`, esto no se registrará correctamente con la compatibilidad CON VERSIONES ANTERIORES. Los consumidores de la nueva versión no podrían leer mensajes antiguos antes de cambiar el esquema, ya que les faltará el campo de `zip code` obligatorio. Sin embargo, si el campo de `zip code` se estableció como opcional en el nuevo esquema, entonces la versión propuesta se registraría correctamente, ya que los consumidores pueden leer el esquema anterior sin el campo de `zip code` opcional. En caso de un caso de uso de gRPC, agregar un nuevo servicio RPC o un método RPC es un cambio compatible con versiones anteriores. Por ejemplo, suponga que tiene una versión de esquema definida por un servicio RPC `MyService` con dos métodos RPC, `Foo` y `Bar`. Si la próxima versión de esquema añade un nuevo método RPC llamado `Baz`, esto se registraría correctamente. Los consumidores podrán leer los datos producidos con el esquema original según la compatibilidad CON VERSIONES ANTERIORES desde el método RPC `Baz` recientemente agregado es opcional. Si tiene una nueva versión de esquema propuesta que agrega un método RPC existente `Foo`, esto no se registrará correctamente con la compatibilidad CON VERSIONES ANTERIORES. Los consumidores de la nueva versión no podrían leer mensajes antiguos antes del cambio de esquema, ya que no pueden entender ni leer datos con el método RPC inexistente `Foo` en una aplicación de gRPC. + *BACKWARD\$1ALL (HACIA ATRÁS\$1TODO)*: esta opción de compatibilidad permite a los consumidores leer tanto la versión actual como todas las versiones anteriores del esquema. Puede utilizar esta opción para comprobar la compatibilidad con la última versión del esquema cuando elimine campos o agregue campos adicionales. + *FORWARD (HACIA ADELANTE)*: esta opción de compatibilidad permite a los consumidores leer tanto la versión actual como la inmediatamente siguiente del esquema, pero no necesariamente versiones posteriores. Puede utilizar esta opción para comprobar la compatibilidad con la última versión del esquema cuando elimine campos o agregue campos adicionales. Un caso de uso típico de HACIA ADELANTE es cuando la aplicación se ha creado para un esquema anterior y debe poder procesar un esquema más reciente. **AVRO** Por ejemplo, supongamos que tiene una versión de esquema definida por nombre (obligatorio), apellido (obligatorio), email (opcional). Si tiene una nueva versión de esquema que agrega un campo obligatorio, p. ej., número de teléfono, esto se registraría correctamente. La compatibilidad HACIA ADELANTE requiere que los consumidores puedan leer los datos producidos con el nuevo esquema utilizando la versión anterior. Si tiene una sugerencia de una versión de esquema que elimina el campo obligatorio de nombre, esto no se registrará correctamente con la compatibilidad HACIA ADELANTE. Los consumidores de la versión anterior no podrían leer los esquemas propuestos, ya que les falta el campo obligatorio de nombre. Sin embargo, si el campo de nombre original era opcional, entonces el nuevo esquema propuesto se registraría correctamente, ya que los consumidores pueden leer datos basados en el nuevo esquema que no incluye el campo opcional de nombre. **JSON** Por ejemplo, supongamos que tiene una versión de esquema definida por nombre (opcional), apellido (opcional), email (opcional) y número de teléfono (opcional). Si tiene una nueva versión de esquema que elimina la propiedad opcional de número de teléfono, esto se registraría correctamente siempre y cuando la nueva versión del esquema no permita ninguna propiedad adicional al configurar el campo `additionalProperties` en falso. La compatibilidad HACIA ADELANTE requiere que los consumidores puedan leer los datos producidos con el nuevo esquema utilizando la versión anterior. Si tiene una sugerencia de una versión de esquema que elimina la propiedad opcional de número de teléfono, esto no se registraría correctamente con la compatibilidad HACIA ADELANTE cuando la nueva versión del esquema establece el campo `additionalProperties` a verdadero, es decir, permite cualquier propiedad adicional. Los consumidores de la versión anterior no podrían leer los esquemas propuestos, ya que podrían tener la propiedad de número de teléfono en un tipo diferente, por ejemplo, cadena en lugar de número. **PROTOBUF** Por ejemplo, suponga que tiene una versión de esquema definida por un mensaje `Person` con `first name` (obligatorio), `last name`(obligatorio) y `email` (opcional) en la sintaxis proto2. De manera semejante a las situaciones de AVRO, si tiene una nueva versión de esquema que agrega un campo obligatorio, p. ej., `phone number`, esto se registraría correctamente. La compatibilidad HACIA ADELANTE requiere que los consumidores puedan leer los datos producidos con el nuevo esquema utilizando la versión anterior. Si tiene una sugerencia de una versión de esquema que elimina el campo obligatorio de `first name`, esto no se registrará correctamente con la compatibilidad CON VERSIONES FUTURAS. Los consumidores de la versión anterior no podrían leer los esquemas propuestos, ya que les falta el campo obligatorio de `first name`. Sin embargo, si el campo de `first name` original era opcional, entonces el nuevo esquema propuesto se registraría correctamente, ya que los consumidores pueden leer datos basados en el nuevo esquema que no incluye el campo opcional de `first name`. En caso de un caso de uso de gRPC, quitar un servicio RPC o un método RPC es un cambio compatible con versiones futuras. Por ejemplo, suponga que tiene una versión de esquema definida por un servicio RPC `MyService` con dos métodos RPC, `Foo` y `Bar`. Si la próxima versión del esquema elimina el método RPC existente denominado `Foo`, esto se registraría correctamente según la compatibilidad CON VERSIONES FUTURAS, ya que los consumidores pueden leer los datos producidos con el nuevo esquema utilizando la versión anterior. Si tiene una sugerencia de una nueva versión de esquema que agrega un método RPC de `Baz`, esto no se registrará correctamente con la compatibilidad CON VERSIONES FUTURAS. Los consumidores de la versión anterior no podrían leer los esquemas propuestos, ya que les falta el método RPC de `Baz`. + *FORWARD\$1ALL (HACIA ADELANTE\$1TODO)*: esta opción de compatibilidad permite a los consumidores leer datos escritos por los productores de cualquier nuevo esquema registrado. Puede utilizar esta opción cuando necesite agregar campos o eliminar campos opcionales, y comprobar la compatibilidad con todas las versiones de esquema anteriores. + *FULL (COMPLETO)*: esta opción de compatibilidad permite a los consumidores leer los datos escritos por los productores con la versión inmediatamente anterior o siguiente del esquema, pero no necesariamente versiones anteriores o posteriores. Puede utilizar esta opción para comprobar la compatibilidad con la última versión del esquema cuando elimine campos o agregue campos adicionales. + *FULL\$1ALL (COMPLETO\$1TODO)*: esta opción de compatibilidad permite a los consumidores leer los datos escritos por los productores con todas las versiones de esquema anteriores. Puede utilizar esta opción para comprobar la compatibilidad con todas las versiones anteriores del esquema, cuando agregue o elimine campos opcionales. ## Bibliotecas Serde de código abierto AWS proporciona bibliotecas Serde de código abierto como marco para serializar y deserializar datos. El diseño de código abierto de estas bibliotecas permite que las aplicaciones y marcos de código abierto comunes soporten estas bibliotecas en sus proyectos. Para obtener más detalles sobre cómo funcionan las bibliotecas de Serde, consulte [Cómo funciona Schema Registry](schema-registry-works.md). ## Cuotas del Schema Registry Las cuotas, también conocidas como límites en AWS, son el valor máximo de los recursos, acciones y elementos de su cuenta de AWS. Los siguientes son los límites flexibles para el Schema Registry en AWS Glue. **Pares de clave-valor de metadatos de la versión del esquema** Puede tener hasta 10 pares clave-valor por SchemaVersion por región de AWS. Puede ver o establecer los pares de metadatos clave-valor con las API [Acción QuerySchemaVersionMetadata (Python: query\$1schema\$1version\$1metadata)](aws-glue-api-schema-registry-api.md#aws-glue-api-schema-registry-api-QuerySchemaVersionMetadata) o [Acción PutSchemaVersionMetadata (Python: put\$1schema\$1version\$1metadata)](aws-glue-api-schema-registry-api.md#aws-glue-api-schema-registry-api-PutSchemaVersionMetadata). Los siguientes son los límites invariables para el Schema Registry en AWS Glue. **Registries** Puede tener hasta 100 registros por región de AWS para esta cuenta. **SchemaVersion** Puede tener hasta 10 000 versiones de esquema por región de AWS para esta cuenta. Cada esquema nuevo crea una nueva versión de esquema, por lo que teóricamente puede tener hasta 10 000 esquemas por cuenta por región, si es que cada esquema tiene una sola versión. **Cargas de esquema** Hay un límite de tamaño de 170 KB para las cargas de esquema. # Cómo funciona Schema Registry En esta sección se describe cómo funcionan los procesos de serialización y deserialización en Schema Registry. 1. Registrar un esquema: si el esquema aún no existe en el registro, puede registrarse con un nombre de esquema igual al nombre del destino (por ejemplo, test\$1topic, test\$1stream, prod\$1firehose) o el productor puede proporcionar un nombre personalizado para el esquema. Los productores también pueden agregar pares clave-valor al esquema como metadatos, como orígenes: msk\$1kafka\$1topic\$1a, o aplicar etiquetas AWS a esquemas en la creación de esquemas. Una vez registrado un esquema, el Schema Registry devuelve el ID de versión del esquema al serializador. Si el esquema existe pero el serializador está utilizando una nueva versión que no existe, Schema Registry comprobará la referencia del esquema en función de una regla de compatibilidad para asegurarse de que la nueva versión es compatible antes de registrarla como una nueva versión. Existen dos métodos para registrar un esquema: registro manual y registro automático. Puede registrar un esquema manualmente a través de la consola de AWS Glue o CLI/SDK. Cuando el registro automático está activado en la configuración del serializador, se realizará el registro automático del esquema. Si no se proporciona el `REGISTRY_NAME` en las configuraciones del productor, entonces el registro automático registrará la nueva versión del esquema según el registro predeterminado (default-registry). Consulte [Instalación de bibliotecas SerDe](schema-registry-gs-serde.md) para obtener información sobre cómo especificar la propiedad de registro automático. 1. El serializador valida los registros de datos respecto del esquema: cuando la aplicación que produce datos ha registrado su esquema, el serializador de Schema Registry valida que el registro que está produciendo la aplicación se estructure con los campos y tipos de datos que coincidan con un esquema registrado. Si el esquema del registro no coincide con un esquema registrado, el serializador devolverá una excepción y la aplicación no entregará el registro al destino. Si no existe ningún esquema y si el nombre del esquema no se proporciona a través de las configuraciones del productor, el esquema se crea con el mismo nombre que el nombre del tema (si es Apache Kafka o Amazon MSK) o el nombre del flujo (si es Kinesis Data Streams). Cada registro tiene una definición de esquema y datos. La definición de esquema se consulta respecto de los esquemas y versiones existentes en el Schema Registry. De forma predeterminada, los productores almacenan en caché las definiciones de esquema y los ID de versión de esquema de los esquemas registrados. Si la definición de versión de esquema de un registro no coincide con lo que está disponible en caché, el productor intentará validar el esquema con Schema Registry. Si la versión del esquema es válida, su ID de versión y definición se almacenarán en caché de manera local en el productor. Puede ajustar el período de caché predeterminado (24 horas) dentro de las propiedades del productor opcionales en el paso n.º 3 de [Instalación de bibliotecas SerDe](schema-registry-gs-serde.md). 1. Serializar y entregar registros: si el registro cumple con el esquema, el serializador agrega a cada registro el ID de la versión del esquema, serializa el registro en función del formato de datos seleccionado (AVRO, JSON, Protobuf u otros formatos próximamente), comprime el registro (configuración opcional del productor) y lo entrega al destino. 1. Los consumidores deserializan los datos: los consumidores que leen estos datos utilizan la biblioteca del deserializador de Schema Registry que analiza el ID de versión del esquema proveniente de la carga del registro. 1. El deserializador puede solicitar el esquema desde el Schema Registry: si es la primera vez que el deserializador ha visto registros con un identificador de versión de esquema concreto, a través del ID de versión de esquema, el deserializador solicitará el esquema desde Schema Registry y almacenará el esquema localmente en el consumidor. Si Schema Registry no puede deserializar el registro, el consumidor puede registrar los datos a partir del registro y continuar, o detener la aplicación. 1. El deserializador utiliza el esquema para deserializar el registro: cuando el deserializador recupera el ID de versión del esquema de Schema Registry, descomprime el registro (si el registro enviado por el productor está comprimido) y utiliza el esquema para deserializar el registro. La aplicación procesa ahora el registro. **nota** Cifrado: sus clientes se comunican con Schema Registry a través de llamadas a la API que cifran los datos en tránsito mediante el cifrado TLS a través de HTTPS. Los esquemas almacenados en Schema Registry siempre se cifran en reposo mediante una clave AWS Key Management Service (AWS KMS) administrada por el servicio. **nota** Autorización de usuario: el registro de esquemas admite políticas de IAM basadas en identidad. # Introducción a Schema Registry Las siguientes secciones ofrecen información general y una guía para configurar y empezar a utilizar Schema Registry. Para obtener más información sobre los conceptos y componentes de Schema Registry, consulte [AWS Glue Schema Registry](schema-registry.md). **Topics** + [Instalación de bibliotecas SerDe](schema-registry-gs-serde.md) + [Integración con AWS Glue Schema Registry](schema-registry-integrations.md) + [Migración desde un registro de esquemas de terceros a AWS Glue Schema Registry](schema-registry-integrations-migration.md) # Instalación de bibliotecas SerDe Las bibliotecas SerDe proporcionan un marco para serializar y deserializar datos. Instalará el serializador de código abierto para sus aplicaciones que producen datos (colectivamente los “serializadores”). El serializador controla la serialización, la compresión y la interacción con Schema Registry. El serializador extrae automáticamente el esquema de un registro que se está escribiendo en un destino compatible con Schema Registry, como Amazon MSK. Del mismo modo, instalará el deserializador de código abierto en sus aplicaciones que consumen datos. # Implementación de Java **nota** Requisitos previos: antes de completar los siguientes pasos, deberá tener un clúster de Amazon Managed Streaming for Apache Kafka (Amazon MSK) o Apache Kafka en ejecución. Sus productores y consumidores deben ejecutarse en Java 8 o superior. Para instalar las bibliotecas en productores y consumidores: 1. Dentro de los archivos pom.xml de los productores y consumidores, agregue esta dependencia a través del siguiente código: ``` software.amazon.glue schema-registry-serde 1.1.5 ``` También puede clonar el [repositorio Github de AWS Glue Schema Registry](https://github.com/awslabs/aws-glue-schema-registry). 1. Configure sus productores con estas propiedades requeridas: ``` 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" ``` Si no hay esquemas existentes, debe activarse el registro automático (siguiente paso). Si tiene un esquema que desea aplicar, reemplace “my-schema (mi esquema)” con su nombre de esquema. También debe proporcionarse el “registry-name (nombre del registro)” si el registro automático del esquema está desactivado. Si el esquema se crea como “default-registry (registro predeterminado)”, entonces el nombre del registro se puede omitir. 1. (Opcional) configure cualquiera de estas propiedades de productor opcionales. Para ver descripciones detalladas de propiedades, consulte [el archivo ReadMe (Léame)](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 ``` El registro automático registra la versión del esquema en el registro predeterminado (“default-registry”). Si no se especifica un `SCHEMA_NAME` en el paso anterior, entonces el nombre del tema se infiere como el `SCHEMA_NAME`. Consulte [Compatibilidad y control de versiones de esquemas](schema-registry.md#schema-registry-compatibility) para obtener más información sobre los modos de compatibilidad. 1. Configure sus consumidores con estas propiedades obligatorias: ``` 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 Región de AWS props.put(AWSSchemaRegistryConstants.AVRO_RECORD_TYPE, AvroRecordType.GENERIC_RECORD.getName()); // Only required for AVRO data format ``` 1. (Opcional) configure estas propiedades de consumidores opcionales. Para ver descripciones detalladas de propiedades, consulte [el archivo ReadMe (Léame)](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 ``` # Implementación de C\$1 **nota** Requisitos previos: antes de completar los siguientes pasos, deberá tener un clúster de Amazon Managed Streaming for Apache Kafka (Amazon MSK) o Apache Kafka en ejecución. Sus productores y consumidores deben ejecutarse en .NET 8.0 o superior. ## Instalación Para las aplicaciones de C\$1, instale el paquete de NuGet de SerDe de registro de esquema de AWS Glue mediante uno de los siguientes métodos: **.NET CLI:** Para instalar el paquete, use el siguiente comando: ``` dotnet add package Aws.Glue.SchemaRegistry --version 1.0.0- ``` donde `` podría ser `1.0.0-linux-x64`, `1.0.0-linux-musl-x64` o `1.0.0-linux-arm64` **PackageReference (en su archivo.csproj):** Añada lo siguiente al archivo del proyecto: ``` ``` donde `` podría ser `1.0.0-linux-x64`, `1.0.0-linux-musl-x64` o `1.0.0-linux-arm64` ## Configuración del archivo de configuración Cree un archivo de propiedades de configuración (por ejemplo, `gsr-config.properties`) con los parámetros necesarios: **Configuración mínima:** El siguiente ejemplo es una configuración mínima de muestra: ``` region=us-east-1 registry.name=default-registry dataFormat=AVRO schemaAutoRegistrationEnabled=true ``` ## Uso de la biblioteca de cliente de esquema de Glue de C\$1 para Kafka SerDes **Ejemplo de uso del serializador:** En el siguiente ejemplo se muestra cómo utilizar el serializador: ``` 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) ``` **Ejemplo de uso del deserializador:** En el siguiente ejemplo se muestra cómo utilizar el deserializador: ``` 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); ``` ## Uso de la biblioteca de cliente de esquema de Glue de C\$1 con KafkaFlow para SerDes **Ejemplo de uso del serializador:** En el siguiente ejemplo se muestra cómo configurar KafkaFlow con el serializador: ``` 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") ) ) ) ) ); ``` **Ejemplo de uso del deserializador:** En el siguiente ejemplo se muestra cómo configurar KafkaFlow con el deserializador: ``` .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()) ) ) ``` ## Propiedades opcionales del productor Puede ampliar el archivo de configuración con propiedades opcionales adicionales: ``` # 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 ``` ## Formatos de fecha admitidos Las implementaciones de Java y C\$1 admiten los mismos formatos de datos: + *AVRO*: formato binario JSON Apache Avro. + *JSON*: formato de esquemas JSON + *PROTOBUF*: formato de búferes de protocolo ## Notas + Para empezar a utilizar la biblioteca, visite [https://www.nuget.org/packages/AWS.Glue.SchemaRegistry](https://www.nuget.org/packages/AWS.Glue.SchemaRegistry) + El código fuente está disponible en: [https://github.com/awslabs/aws-glue-schema-registry](https://github.com/awslabs/aws-glue-schema-registry) # Creación de un registro Puede utilizar el registro predeterminado o crear tantos registros nuevos como sea necesario mediante las API de AWS Glue o la consola de AWS Glue. **AWS GlueLas API de** Puede seguir estos pasos para realizar esta tarea mediante las API de AWS Glue. Para utilizar AWS CLI para las API de AWS Glue Schema Registry, asegúrese de actualizar su AWS CLI a la versión más reciente. Para agregar un registro nuevo, use la API de [Acción CreateRegistry (Python: create\$1registry)](aws-glue-api-schema-registry-api.md#aws-glue-api-schema-registry-api-CreateRegistry). Especifique `RegistryName` como el nombre del registro que se va a crear con una longitud máxima de 255, y solo puede contener letras, números, guión, guión bajo, signo de dólar o marca hash. Especifique una `Description` como una cadena cuya extensión no sea más de 2048 bytes y que coincida con el [patrón de cadena de varias líneas de la dirección URI](https://docs.aws.amazon.com/glue/latest/dg/aws-glue-api-common.html#aws-glue-api-common-_string-patterns). De manera opcional, especifique una o varias `Tags` para su registro, como matriz de mapas de pares clave-valor. ``` aws glue create-registry --registry-name registryName1 --description description ``` Cuando se crea el registro se le asigna un nombre de recurso de Amazon (ARN). Este ARN lo puede ver en `RegistryArn` de la respuesta de la API. Ahora que ha creado un registro, cree uno o más esquemas para ese registro. **Consola de AWS Glue** Para agregar un nuevo registro en la consola de AWS Glue: 1. Inicie sesión en Consola de administración de AWS y abra la consola de AWS Glue en [https://console.aws.amazon.com/glue/](https://console.aws.amazon.com/glue\). 1. En el panel de navegación, en **Data catalog** (Catálogo de datos), elija **Schema registries (Registros de esquemas)**. 1. Elija **Add registry (Agregar registro)**. 1. Escriba un **Registry name (Nombre del Registro)** para el registro, que conste de letras, números, guiones o guiones bajos. Este nombre no se puede cambiar. 1. Escriba una **Description (Descripción)** (opcional) para el registro. 1. De manera opcional, aplique una o varias etiquetas a su registro. Elija **Add new tag (Agregar nueva etiqueta)** y especifique una **Tag key (Clave de etiqueta)** y, opcionalmente, un **Tag value (Valor de etiqueta)**. 1. Elija **Add registry (Agregar registro)**. ![\[Ejemplo de creación de un registro.\]](http://docs.aws.amazon.com/es_es/glue/latest/dg/images/schema_reg_create_registry.png) Cuando se crea el registro, se le asigna un nombre de recurso de Amazon (ARN), que puede ver al seleccionar el registro de la lista en **Schema registries (Registros de esquemas)**. Ahora que ha creado un registro, cree uno o más esquemas para ese registro. # Trabajar con un registro específico (JAVA POJO) para JSON Puede usar un objeto Java antiguo simple (POJO) y transferir el objeto como un registro. Esto es similar a la noción de un registro específico en AVRO. [mbknor-jackson-jsonschema](https://github.com/mbknor/mbknor-jackson-jsonSchema) puede generar un esquema JSON para el POJO transferido. Esta biblioteca también puede inyectar información adicional en el esquema JSON. La biblioteca de AWS Glue Schema Registry utiliza el campo “className (nombre de clase)” inyectado en el esquema para proporcionar un nombre de clase completamente clasificado. El deserializador utiliza el campo “className” para deserializar en un objeto de esa clase. ``` 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() {} } ``` # Creación de un esquema Puede crear un esquema con las API de AWS Glue o la consola de AWS Glue. **AWS GlueLas API de** Puede seguir estos pasos para realizar esta tarea mediante las API de AWS Glue. Para agregar un esquema nuevo, use la API de [Acción CreateSchema (Python: create\$1schema)](aws-glue-api-schema-registry-api.md#aws-glue-api-schema-registry-api-CreateSchema). Especifique una estructura de `RegistryId` para indicar un registro para el esquema. O bien, omita el `RegistryId` para utilizar el registro predeterminado. Especifique un `SchemaName`, que conste de letras, números, guiones o guiones bajos, y `DataFormat` como **AVRO** o **JSON**. Una vez configurado el `DataFormat` en un esquema no puede modificarse. Especifique un modo de `Compatibility`: + *Hacia atrás (recomendado)*: el consumidor puede leer tanto la versión actual como la anterior. + *Hacia atrás todo*: el consumidor puede leer tanto la versión actual como todas las anteriores. + *Hacia adelante*: el consumidor puede leer tanto la versión actual como la posterior. + *Hacia adelante todo*: el consumidor puede leer tanto la versión actual como todas las versiones posteriores. + *Completo*: combinación de hacia atrás y hacia adelante. + *Completo todo*: combinación de hacia atrás todos y hacia adelante todos. + *Ninguno*: no se realizan comprobaciones de compatibilidad. + *Desactivado*: evita cualquier control de versiones para este esquema. Opcionalmente, especifique `Tags` para su esquema. Especifique una `SchemaDefinition` para definir el esquema en formato de datos Avro, JSON o Protobuf. Consulte estos ejemplos. Para el formato de datos Avro: ``` 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\"} ]}" ``` Para el formato de datos JSON: ``` 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\"}}}" ``` Para el formato de datos Protobuf: ``` 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;}" ``` **Consola de AWS Glue** Para agregar un nuevo esquema a través de la consola de AWS Glue: 1. Inicie sesión en la consola de administración de AWS y abra la consola de AWS Glue en [https://console.aws.amazon.com/glue/](https://console.aws.amazon.com/glue\). 1. En el panel de navegación, en **Data catalog** (Catálogo de datos), elija **Schema (Esquemas)**. 1. Elija **Add schema (Agregar esquema)**. 1. Escriba un **Schema name (Nombre del esquema)**, que conste de letras, números, guiones, guiones bajos, signos de dólar o marcas hash. Este nombre no se puede cambiar. 1. Elija el **Registry (Registro)** en el que se almacenará el esquema del menú desplegable. El registro principal no se puede cambiar después de la creación. 1. Deje el **Data format (Formato de los datos)** como *Apache Avro* o *JSON*. Este formato se aplica a todas las versiones de este esquema. 1. Elija un **(Compatibility mode) Modo de compatibilidad**. + *Hacia atrás (recomendado)*: el receptor puede leer tanto la versión actual como la anterior. + *Hacia atrás todo*: el receptor puede leer tanto la versión actual como todas las anteriores. + *Hacia adelante*: el remitente puede escribir tanto la versión actual como la anterior. + *Hacia adelante todo*: el remitente puede escribir tanto la versión actual como todas las versiones anteriores. + *Completo*: combinación de hacia atrás y hacia adelante. + *Completo todo*: combinación de hacia atrás todos y hacia adelante todos. + *Ninguno*: no se realizan comprobaciones de compatibilidad. + *Desactivado*: evita cualquier control de versiones para este esquema. 1. Escriba una **Description (Descripción)** para el registro de hasta 250 caracteres. ![\[Ejemplo de creación de un esquema.\]](http://docs.aws.amazon.com/es_es/glue/latest/dg/images/schema_reg_create_schema.png) 1. De manera opcional, aplique una o varias etiquetas a su esquema. Elija **Add new tag (Agregar nueva etiqueta)** y especifique una **Tag key (Clave de etiqueta)** y, opcionalmente, un **Tag value (Valor de etiqueta)**. 1. En la casilla, **First schema version (Primera versión del esquema)**, ingrese o pegue el esquema inicial. Para el formato Avro, consulte [Trabajar con formato de datos Avro](#schema-registry-avro) Para el formato JSON, consulte [Trabajar con formato de datos JSON](#schema-registry-json) 1. También puede seleccionar **Add metadata (Agregar metadatos)** para agregar metadatos de versión para anotar o clasificar la versión del esquema. 1. Elija **Create schema and version (Crear esquema y versión)**. ![\[Ejemplo de creación de un esquema.\]](http://docs.aws.amazon.com/es_es/glue/latest/dg/images/schema_reg_create_schema2.png) El esquema se crea y aparece en la lista en **Schemas (Esquemas)**. ## Trabajar con formato de datos Avro Avro proporciona servicios de serialización e intercambio de datos. Avro almacena la definición de datos en formato JSON, lo que facilita su lectura e interpretación. Los datos en sí se almacenan en formato binario. Para obtener información sobre cómo definir un esquema Apache Avro, consulte la [especificación de Apache Avro](http://avro.apache.org/docs/current/spec.html). ## Trabajar con formato de datos JSON Los datos se pueden serializar con formato JSON. [Formato de esquemas JSON](https://json-schema.org/) define el estándar para el formato de esquema JSON. # Actualización de un esquema o registro Una vez creado, puede editar los esquemas, las versiones de esquema o el registro. ## Actualización de un registro Puede actualizar un registro mediante las API de AWS Glue o la consola de AWS Glue. No se puede editar el nombre de un registro existente. La descripción de un registro puede editarse. **AWS GlueLas API de** Para actualizar un registro existente, use la API [Acción UpdateRegistry (Python: update\$1registry)](aws-glue-api-schema-registry-api.md#aws-glue-api-schema-registry-api-UpdateRegistry). Especifique una estructura de `RegistryId` para indicar el registro que desea actualizar. Transferir una `Description` para cambiar la descripción de un registro. ``` aws glue update-registry --description updatedDescription --registry-id RegistryArn="arn:aws:glue:us-east-2:901234567890:registry/registryName1" ``` **Consola de AWS Glue** Para actualizar un registro mediante la consola de AWS Glue: 1. Inicie sesión en Consola de administración de AWS y abra la consola de AWS Glue en [https://console.aws.amazon.com/glue/](https://console.aws.amazon.com/glue\). 1. En el panel de navegación, en **Data catalog** (Catálogo de datos), elija **Schema registries (Registros de esquemas)**. 1. Seleccione un registro de la lista de registros, al marcar la casilla correspondiente. 1. En el menú **Action (Acción)**, elija **Edit registry (Editar registro)**. # Actualización de un esquema Puede actualizar la descripción o la configuración de compatibilidad de un esquema. Para actualizar un esquema existente, use la API [Acción UpdateSchema (Python: update\$1schema)](aws-glue-api-schema-registry-api.md#aws-glue-api-schema-registry-api-UpdateSchema). Especifique una estructura de `SchemaId` para indicar el esquema que desea actualizar. Uno de `VersionNumber` o `Compatibility` debe ser proporcionado. Ejemplos de código 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 ``` # Agregar una versión de esquema. Cuando agregue una versión de esquema, deberá comparar las versiones para asegurarse de que se aceptará el nuevo esquema. Para agregar una nueva versión a un esquema existente, utilice la API [Acción RegisterSchemaVersion (Python: register\$1schema\$1version)](aws-glue-api-schema-registry-api.md#aws-glue-api-schema-registry-api-RegisterSchemaVersion). Especifique una estructura de `SchemaId` para indicar el esquema para el que desea agregar una versión, y una `SchemaDefinition` para definir el esquema. Ejemplos de código 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. Inicie sesión en Consola de administración de AWS y abra la consola de AWS Glue en [https://console.aws.amazon.com/glue/](https://console.aws.amazon.com/glue\). 1. En el panel de navegación, en **Data catalog** (Catálogo de datos), elija **Schema (Esquemas)**. 1. Seleccione el esquema de la lista de esquemas, al marcar la casilla correspondiente. 1. Seleccione uno o varios esquemas de la lista, al marcar las casillas correspondientes. 1. En el menú **Action (Acción)**, elija **Register new version (Registrar nueva versión)**. 1. En la casilla **New version (Nueva versión)**, ingrese o pegue el nuevo esquema. 1. Elija **Compare with previous version (Comparar con la versión anterior)** para ver las diferencias con la versión del esquema anterior. 1. También puede seleccionar **Add metadata (Agregar metadatos)** para agregar metadatos de versión para anotar o clasificar la versión del esquema. Ingrese **Key (Clave)** y **Value (Valor)** opcional. 1. Elija **Register version (Registrar versión)**. ![\[Agregar una versión de esquema.\]](http://docs.aws.amazon.com/es_es/glue/latest/dg/images/schema_reg_add_schema_version.png) La versión de esquema(s) aparece en la lista de versiones. Si la versión cambió el modo de compatibilidad, la versión se marcará como punto de control. ## Ejemplo de comparación de versiones de esquema. Cuando elige **Compare with previous version (Comparar con la versión anterior)**, verá las versiones anterior y nueva mostradas juntas. La información que se modificó se resaltará de la siguiente manera: + *Amarillo*: indica información modificada. + *Verde*: indica el contenido agregado en la última versión. + *Rojo*: indica el contenido eliminado de la última versión. También se pueden realizar comparaciones con versiones anteriores. ![\[Ejemplo de comparación de versiones de esquema.\]](http://docs.aws.amazon.com/es_es/glue/latest/dg/images/schema_reg_version_comparison.png) # Eliminación de un esquema o registro Eliminar un esquema, una versión de esquema o un registro son acciones permanentes que no se pueden deshacer. ## Eliminación de un esquema Es posible que desee eliminar un esquema cuando ya no se utilizará dentro de un registro, utilizando la consola Consola de administración de AWS, o la API [Acción DeleteSchema (Python: delete\$1schema)](aws-glue-api-schema-registry-api.md#aws-glue-api-schema-registry-api-DeleteSchema). Eliminar uno o varios esquemas es una acción permanente que no se puede deshacer. Asegúrese de que el esquema o los esquemas ya no son necesarios. Para eliminar un esquema del registro, llame a la API [Acción DeleteSchema (Python: delete\$1schema)](aws-glue-api-schema-registry-api.md#aws-glue-api-schema-registry-api-DeleteSchema), y especifique la estructura de `SchemaId` para identificar el esquema. Por ejemplo: ``` 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" ``` **Consola de AWS Glue** Para eliminar un esquema de la consola de AWS Glue: 1. Inicie sesión en Consola de administración de AWS y abra la consola de AWS Glue en [https://console.aws.amazon.com/glue/](https://console.aws.amazon.com/glue\). 1. En el panel de navegación, en **Data catalog** (Catálogo de datos), elija **Schema registries (Registros de esquemas)**. 1. Elija el registro que contiene el esquema de la lista de registros. 1. Seleccione uno o varios esquemas de la lista, al marcar las casillas correspondientes. 1. En el menú **Action (Acción)**, elija **Delete schema (Eliminar esquema)**. 1. Ingrese el texto **Delete** en el campo para confirmar la eliminación. 1. Elija **Eliminar**. Los esquemas especificados se eliminan del registro. ## Eliminar una versión de esquema A medida que los esquemas se acumulan en el registro, es posible que desee eliminar las versiones de esquema no deseadas mediante la Consola de administración de AWS o la API [Acción DeleteSchemaVersions (Python: delete\$1schema\$1versions)](aws-glue-api-schema-registry-api.md#aws-glue-api-schema-registry-api-DeleteSchemaVersions). Eliminar una o varias versiones de esquema es una acción permanente que no se puede deshacer. Asegúrese de que las versiones del esquema ya no sean necesarias. Al eliminar versiones de esquema, tenga en cuenta las siguientes restricciones: + No puede eliminar una versión marcada como punto de control. + El rango de versiones contiguas no puede ser superior a 25. + La versión del esquema más reciente no debe estar en un estado pendiente. Especifique la estructura de `SchemaId` para identificar el esquema y especifique las `Versions` como un rango de versiones para eliminar. Para obtener más información sobre cómo especificar una versión o un rango de versiones, consulte [Acción DeleteRegistry (Python: delete\$1registry)](aws-glue-api-schema-registry-api.md#aws-glue-api-schema-registry-api-DeleteRegistry). Las versiones de los esquemas especificados se eliminan del registro. Llamar a la API [Acción ListSchemaVersions (Python: list\$1schema\$1versions)](aws-glue-api-schema-registry-api.md#aws-glue-api-schema-registry-api-ListSchemaVersions) después de esta llamada arrojará una lista del estado de las versiones eliminadas. Por ejemplo: ``` 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. Inicie sesión en Consola de administración de AWS y abra la consola de AWS Glue en [https://console.aws.amazon.com/glue/](https://console.aws.amazon.com/glue\). 1. En el panel de navegación, en **Data catalog** (Catálogo de datos), elija **Schema registries (Registros de esquemas)**. 1. Elija el registro que contiene el esquema de la lista de registros. 1. Seleccione uno o varios esquemas de la lista, al marcar las casillas correspondientes. 1. En el menú **Action (Acción)**, elija **Delete schema (Eliminar esquema)**. 1. Ingrese el texto **Delete** en el campo para confirmar la eliminación. 1. Elija **Eliminar**. Las versiones de los esquemas especificados se eliminan del registro. # Eliminación de un registro Es posible que desee eliminar un registro cuando los esquemas que contiene ya no se organicen en ese registro. Tendrá que reasignar esos esquemas a otro registro. Eliminar uno o varios registros es una acción permanente que no se puede deshacer. Asegúrese de que el registro o los registros ya no son necesarios. El registro predeterminado se puede eliminar mediante AWS CLI. **AWS GlueAPI de** Para eliminar todo el registro, incluido el esquema y todas sus versiones, llame a la API [Acción DeleteRegistry (Python: delete\$1registry)](aws-glue-api-schema-registry-api.md#aws-glue-api-schema-registry-api-DeleteRegistry). Especifique una estructura de `RegistryId` para identificar el registro. Por ejemplo: ``` 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" ``` Para obtener el estado de la operación de eliminación, puede llamar a la API `GetRegistry` después de la llamada asíncrona. **Consola de AWS Glue** Para eliminar un registro de la consola de AWS Glue: 1. Inicie sesión en Consola de administración de AWS y abra la consola de AWS Glue en [https://console.aws.amazon.com/glue/](https://console.aws.amazon.com/glue\). 1. En el panel de navegación, en **Data catalog** (Catálogo de datos), elija **Schema registries (Registros de esquemas)**. 1. Seleccione un registro de la lista, al marcar la casilla correspondiente. 1. En el menú **Action (Acción)**, elija **Delete registry (Eliminar registro)**. 1. Ingrese el texto **Delete** en el campo para confirmar la eliminación. 1. Elija **Eliminar**. Los registros seleccionados se eliminan de AWS Glue. ## Ejemplos de IAM para serializadores **nota** AWSLas políticas administradas de conceden los permisos necesarios para los casos de uso comunes. Para obtener información sobre el uso de las políticas para administrar el registro de esquemas, consulte [AWS Políticas administradas (predefinidas) por para AWS Glue](security-iam-awsmanpol.md#access-policy-examples-aws-managed). Para los serializadores, debe crear una política mínima similar a la siguiente para poder encontrar el `schemaVersionId` para una definición de esquema determinada. Tenga en cuenta que debe tener permisos de lectura en el registro para poder leer los esquemas del registro. Puede limitar los registros que se pueden leer mediante la cláusula `Resource`. Ejemplos de código 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" ] } ``` Además, también puede permitir a los productores crear nuevos esquemas y versiones mediante la inclusión de los siguientes métodos adicionales. Tenga en cuenta que debería poder inspeccionar el registro para agregar/eliminar/evolucionar los esquemas dentro de él. Puede limitar los registros que se pueden inspeccionar mediante la cláusula `Resource`. Ejemplos de código 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" ] } ``` ## Ejemplos de IAM para deserializadores Para los deserializadores (lado del consumidor), debe crear una política similar a la siguiente para permitir que el deserializador recupere el esquema de Schema Registry para su deserialización. Tenga en cuenta que debería poder inspeccionar el registro para obtener los esquemas dentro de él. Ejemplos de código 15: ``` { "Sid" : "GetSchemaVersion", "Effect" : "Allow", "Action" : [ "glue:GetSchemaVersion" ], "Resource" : ["*"] } ``` ## Conectividad privada mediante AWS PrivateLink Puede usar AWS PrivateLink para conectar la VPC del productor de datos a AWS Glue. Para ello, defina un punto de enlace de la VPC de interfaz para AWS Glue. Cuando se utiliza un punto de enlace de la interfaz de la VPC, la comunicación entre la VPC y AWS Glue se realiza en su totalidad dentro de la red de AWS. Para obtener más información, consulte [Uso de AWS Glue con puntos de enlace de la VPC](https://docs.aws.amazon.com/glue/latest/dg/vpc-endpoint.html). # Acceso a métricas de Amazon CloudWatch Las métricas de Amazon CloudWatch están disponibles como parte de la capa gratuita de CloudWatch. También puede acceder a estas métricas en la consola de CloudWatch. Las métricas de nivel de API incluyen CreateSchema (éxito y latencia), GetSchemabyDefinition, (éxito y latencia), GetSchemaVersion (éxito y latencia), RegisterSchemaVersion (éxito y latencia), PutSchemaVersionMetadata (éxito y latencia). Las métricas de nivel de recursos incluyen Registry.ThrottledByLimit, SchemaVersion.ThrottledByLimit, SchemaVersion.Size. # Ejemplo de plantilla de CloudFormation para Schema Registry A continuación se muestra una plantilla de ejemplo para crear recursos de Schema Registry en CloudFormation. Para crear esta pila en su cuenta, copie la plantilla anterior en un archivo `SampleTemplate.yaml` y ejecute el siguiente comando: ``` aws cloudformation create-stack --stack-name ABCSchemaRegistryStack --template-body "'cat SampleTemplate.yaml'" ``` En este ejemplo se utiliza `AWS::Glue::Registry` para crear un registro, `AWS::Glue::Schema` para crear un esquema, `AWS::Glue::SchemaVersion` para crear una versión de esquema y `AWS::Glue::SchemaVersionMetadata` para completar los metadatos de la versión del esquema. ``` 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" ``` # Integración con AWS Glue Schema Registry En estas secciones se describen las integraciones con AWS Glue Schema Registry. Los ejemplos de esta sección muestran un esquema con formato de datos AVRO. Para obtener más ejemplos, incluidos esquemas con formato de datos JSON, consulte las pruebas de integración y la información de ReadMe (Léame) en el [Repositorio de código abierto de AWS Glue Schema Registry](https://github.com/awslabs/aws-glue-schema-registry). **Topics** + [Caso de uso: Conexión de Schema Registry a Amazon MSK o Apache Kafka](#schema-registry-integrations-amazon-msk) + [Caso de uso: Integración de Amazon Kinesis Data Streams con AWS Glue Schema Registry](#schema-registry-integrations-kds) + [Caso de uso: Amazon Managed Service para Apache Flink](#schema-registry-integrations-kinesis-data-analytics-apache-flink) + [Caso de uso: integración con AWS Lambda](#schema-registry-integrations-aws-lambda) + [Caso de uso: AWS Glue Data Catalog](#schema-registry-integrations-aws-glue-data-catalog) + [Caso de uso: Streaming de AWS Glue](#schema-registry-integrations-aws-glue-streaming) + [Caso de uso: Apache Kafka Streams](#schema-registry-integrations-apache-kafka-streams) ## Caso de uso: Conexión de Schema Registry a Amazon MSK o Apache Kafka Supongamos que está escribiendo datos en un tema de Apache Kafka. Puede seguir estos pasos para comenzar. 1. Cree un clúster de Amazon Managed Streaming for Apache Kafka (Amazon MSK) o Apache Kafka con al menos un tema. Si crea un clúster de Amazon MSK, puede utilizar la Consola de administración de AWS. Siga las siguientes isntrucciones: [Introducción al uso de Amazon MSK](https://docs.aws.amazon.com/msk/latest/developerguide/getting-started.html) en la *Guía para desarrolladores de Amazon Managed Streaming for Apache Kafka*. 1. Siga el paso [Instalación de bibliotecas SerDe](schema-registry-gs-serde.md) anterior. 1. Para crear registros de esquema, esquemas o versiones de esquema, siga las instrucciones de la sección [Introducción a Schema Registry](schema-registry-gs.md) de este documento. 1. Inicie a sus productores y consumidores en el uso de Schema Registry para escribir y leer registros a/desde el tema de Amazon MSK o Apache Kafka. Puede encontrar un ejemplo de código de productor y consumidor en [el archivo ReadMe (Léame)](https://github.com/awslabs/aws-glue-schema-registry/blob/master/README.md) de las bibliotecas Serde. La biblioteca de Schema Registry del productor serializará automáticamente el registro y agregará un ID de versión de esquema al registro. 1. Si se ha introducido el esquema de este registro, o si el registro automático está activado, el esquema se habrá registrado en Schema Registry. 1. El consumidor que lee el tema de Amazon MSK o Apache Kafka, con la biblioteca de AWS Glue Schema Registry, buscará automáticamente el esquema desde Schema Registry. ## Caso de uso: Integración de Amazon Kinesis Data Streams con AWS Glue Schema Registry Esta integración requiere que tenga un flujo de datos de Amazon Kinesis. Para obtener más información, consulte [Introducción a Amazon Kinesis Data Streams](https://docs.aws.amazon.com/streams/latest/dev/getting-started.html) en la *Guía para desarrolladores de Amazon Kinesis Data Streams*. Existen dos formas de interactuar con los datos en un flujo de datos de Kinesis. + A través de las bibliotecas Kinesis Producer Library (KPL) y Kinesis Client Library (KCL) en Java. No se proporciona soporte multilingüe. + A través de las API `PutRecords`, `PutRecord` y `GetRecords` de Kinesis Data Streams disponibles en AWS SDK para Java. Si utiliza actualmente las bibliotecas KPL/KCL, le recomendamos seguir utilizando ese método. Hay versiones actualizadas de KCL y KPL con Schema Registry integrado, como se muestra en los ejemplos. De lo contrario, puede utilizar el código de muestra para aprovechar el AWS Glue Schema Registry si utiliza las API de KDS directamente. La integración de Schema Registry sólo está disponible con KPL v0.14.2 o posterior y con KCL v2.3 o posterior. La integración de Schema Registry con datos JSON sólo está disponible con KPL v0.14.8 o posterior y con KCL v2.3.6 o posterior. ### Interacción con datos mediante SDK de Kinesis V2 En esta sección se describe la interacción con Kinesis mediante SDK de Kinesis 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 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 En esta sección se describe la integración de Kinesis Data Streams con Schema Registry mediante las bibliotecas KPL/KCL. Para obtener más información sobre el uso de KPL/KCL, consulte [Desarrollar productores con Amazon Kinesis Producer Library](https://docs.aws.amazon.com/streams/latest/dev/developing-producers-with-kpl.html) en la *Guía para desarrolladores de Amazon Kinesis Data Streams*. #### Configuración de Schema Registry en KPL 1. Establezca la definición de esquema para los datos, el formato de datos y el nombre del esquema creados en AWS Glue Schema Registry. 1. Opcionalmente, configure el objeto `GlueSchemaRegistryConfiguration`. 1. Transfiera el objeto de esquema a `addUserRecord API`. ``` private static final String SCHEMA_DEFINITION = "{"namespace": "example.avro",\n" + " "type": "record",\n" + " "name": "User",\n" + " "fields": [\n" + " {"name": "name", "type": "string"},\n" + " {"name": "favorite_number", "type": ["int", "null"]},\n" + " {"name": "favorite_color", "type": ["string", "null"]}\n" + " ]\n" + "}"; KinesisProducerConfiguration config = new KinesisProducerConfiguration(); config.setRegion("us-west-1") //[Optional] configuration for Schema Registry. GlueSchemaRegistryConfiguration schemaRegistryConfig = new GlueSchemaRegistryConfiguration("us-west-1"); schemaRegistryConfig.setCompression(true); config.setGlueSchemaRegistryConfiguration(schemaRegistryConfig); ///Optional configuration ends. final KinesisProducer producer = new KinesisProducer(config); final ByteBuffer data = getDataToSend(); com.amazonaws.services.schemaregistry.common.Schema gsrSchema = new Schema(SCHEMA_DEFINITION, DataFormat.AVRO.toString(), "demoSchema"); ListenableFuture 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 Desarrolle un consumidor de Kinesis Client Library en Java. Para obtener más información, consulte [Desarrollo de un consumidor de Kinesis Client Library en Java](https://docs.aws.amazon.com/streams/latest/dev/kcl2-standard-consumer-java-example.html) en la *Guía para desarrolladores de Amazon Kinesis Data Streams*. 1. Cree una instancia de `GlueSchemaRegistryDeserializer` al transferir un objeto `GlueSchemaRegistryConfiguration`. 1. Transfiera el `GlueSchemaRegistryDeserializer` a `retrievalConfig.glueSchemaRegistryDeserializer`. 1. Acceda al esquema de los mensajes entrantes al llamar a `kinesisClientRecord.getSchema()`. ``` GlueSchemaRegistryConfiguration schemaRegistryConfig = new GlueSchemaRegistryConfiguration(this.region.toString()); GlueSchemaRegistryDeserializer glueSchemaRegistryDeserializer = new GlueSchemaRegistryDeserializerImpl(DefaultCredentialsProvider.builder().build(), schemaRegistryConfig); RetrievalConfig retrievalConfig = configsBuilder.retrievalConfig().retrievalSpecificConfig(new PollingConfig(streamName, kinesisClient)); retrievalConfig.glueSchemaRegistryDeserializer(glueSchemaRegistryDeserializer); Scheduler scheduler = new Scheduler( configsBuilder.checkpointConfig(), configsBuilder.coordinatorConfig(), configsBuilder.leaseManagementConfig(), configsBuilder.lifecycleConfig(), configsBuilder.metricsConfig(), configsBuilder.processorConfig(), retrievalConfig ); public void processRecords(ProcessRecordsInput processRecordsInput) { MDC.put(SHARD_ID_MDC_KEY, shardId); try { log.info("Processing {} record(s)", processRecordsInput.records().size()); processRecordsInput.records() .forEach( r -> log.info("Processed record pk: {} -- Seq: {} : data {} with schema: {}", r.partitionKey(), r.sequenceNumber(), recordToAvroObj(r).toString(), r.getSchema())); } catch (Throwable t) { log.error("Caught throwable while processing records. Aborting."); Runtime.getRuntime().halt(1); } finally { MDC.remove(SHARD_ID_MDC_KEY); } } private GenericRecord recordToAvroObj(KinesisClientRecord r) { byte[] data = new byte[r.data().remaining()]; r.data().get(data, 0, data.length); org.apache.avro.Schema schema = new org.apache.avro.Schema.Parser().parse(r.schema().getSchemaDefinition()); DatumReader datumReader = new GenericDatumReader<>(schema); BinaryDecoder binaryDecoder = DecoderFactory.get().binaryDecoder(data, 0, data.length, null); return (GenericRecord) datumReader.read(null, binaryDecoder); } ``` #### Interacción con datos mediante las API de Kinesis Data Streams En esta sección se describe la integración de Kinesis Data Streams con Schema Registry mediante las API de Kinesis Data Streams. 1. Actualice estas dependencias de Maven: ``` 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 El siguiente es el código de ejemplo para usar las API `PutRecords` y `GetRecords`. ``` //Full sample code import com.amazonaws.services.schemaregistry.deserializers.GlueSchemaRegistryDeserializerImpl; import com.amazonaws.services.schemaregistry.serializers.GlueSchemaRegistrySerializerImpl; import com.amazonaws.services.schemaregistry.utils.AVROUtils; import com.amazonaws.services.schemaregistry.utils.AWSSchemaRegistryConstants; import org.apache.avro.Schema; import org.apache.avro.generic.GenericData; import org.apache.avro.generic.GenericDatumReader; import org.apache.avro.generic.GenericDatumWriter; import org.apache.avro.generic.GenericRecord; import org.apache.avro.io.Decoder; import org.apache.avro.io.DecoderFactory; import org.apache.avro.io.Encoder; import org.apache.avro.io.EncoderFactory; import software.amazon.awssdk.auth.credentials.DefaultCredentialsProvider; import software.amazon.awssdk.services.glue.model.DataFormat; import java.io.ByteArrayOutputStream; import java.io.File; import java.io.IOException; import java.nio.ByteBuffer; import java.util.Collections; import java.util.HashMap; import java.util.Map; public class PutAndGetExampleWithEncodedData { static final String regionName = "us-east-2"; static final String streamName = "testStream1"; static final String schemaName = "User-Topic"; static final String AVRO_USER_SCHEMA_FILE = "src/main/resources/user.avsc"; KinesisApi kinesisApi = new KinesisApi(); void runSampleForPutRecord() throws IOException { Object testRecord = getTestRecord(); byte[] recordAsBytes = convertRecordToBytes(testRecord); String schemaDefinition = AVROUtils.getInstance().getSchemaDefinition(testRecord); //The following lines add a Schema Header to a 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(regionName)); byte[] recordWithSchemaHeader = glueSchemaRegistrySerializer.encode(streamName, awsSchema, recordAsBytes); //Use PutRecords api to pass a list of records kinesisApi.putRecords(Collections.singletonList(recordWithSchemaHeader), streamName, regionName); //OR //Use PutRecord api to pass single record //kinesisApi.putRecord(recordWithSchemaHeader, streamName, regionName); } byte[] runSampleForGetRecord() throws IOException { ByteBuffer recordWithSchemaHeader = kinesisApi.getRecords(streamName, regionName); //The following lines remove the schema registry header GlueSchemaRegistryDeserializerImpl glueSchemaRegistryDeserializer = new GlueSchemaRegistryDeserializerImpl(DefaultCredentialsProvider.builder().build(), new GlueSchemaRegistryConfiguration(regionName)); 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); } return record; } private byte[] convertRecordToBytes(final Object record) throws IOException { ByteArrayOutputStream recordAsBytes = new ByteArrayOutputStream(); Encoder encoder = EncoderFactory.get().directBinaryEncoder(recordAsBytes, null); GenericDatumWriter datumWriter = new GenericDatumWriter<>(AVROUtils.getInstance().getSchema(record)); datumWriter.write(record, encoder); encoder.flush(); return recordAsBytes.toByteArray(); } private GenericRecord convertBytesToRecord(Schema avroSchema, byte[] record) throws IOException { final GenericDatumReader 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 Apache Flink es un marco de código abierto y motor de procesamiento distribuido popular para informática con estado sobre flujos de datos ilimitados y delimitados. Amazon Managed Service para Apache Flink es un servicio de AWS completamente administrado que permite crear y administrar aplicaciones de Apache Flink para procesar datos de streaming. El código abierto Apache Flink proporciona una serie de orígenes y receptores. Por ejemplo, los orígenes de datos predefinidos incluyen la lectura de archivos, directorios y sockets, y la ingesta de datos de recopilaciones e iteradores. Los conectores Apache Flink DataStream proporcionan código para que Apache Flink interactúe con varios sistemas de terceros, como Apache Kafka o Kinesis como orígenes o receptores. Para obtener más información, consulte la [Guía para desarrolladores de Amazon Kinesis Data Analytics](https://docs.aws.amazon.com/kinesisanalytics/latest/java/what-is.html). ### Conector Kafka de Apache Flink Apache Flink proporciona un conector de flujo de datos Apache Kafka para leer y escribir datos en temas de Kafka con garantías de una sola vez. El consumidor Kafka de Flink, `FlinkKafkaConsumer`, proporciona acceso a la lectura de uno o más temas de Kafka. El productor Kafka de Apache Flink, `FlinkKafkaProducer`, permite escribir una secuencia de registros en uno o más temas de Kafka. Para obtener más información, consulte [Conector de Apache Kafka](https://ci.apache.org/projects/flink/flink-docs-stable/dev/connectors/kafka.html). ### Conector de flujos de Kinesis de Apache Flink El conector de flujo de datos de Kinesis proporciona acceso a Amazon Kinesis Data Streams. El `FlinkKinesisConsumer` es un origen de datos de streaming en paralelo de exactamente una única vez que se suscribe a múltiples flujos de Kinesis dentro de la misma región de servicio de AWS, y puede manejar de forma transparente la redistribución de flujos mientras el trabajo se está ejecutando. Cada subtarea del consumidor es responsable de obtener registros de datos de múltiples fragmentos de Kinesis. El número de fragmentos obtenidos por cada subtarea cambiará a medida que Kinesis cierre y cree fragmentos. El `FlinkKinesisProducer` utiliza Kinesis Producer Library (KPL) para poner los datos de un flujo de Apache Flink en un flujo de Kinesis. Para obtener más información, consulte [Conector de Amazon Kinesis Streams](https://ci.apache.org/projects/flink/flink-docs-release-1.11/dev/connectors/kinesis.html). Para obtener más información, consulte el [repositorio GitHub de esquemas de AWS Glue](https://github.com/awslabs/aws-glue-schema-registry). ### Integración con Apache Flink La biblioteca SerDes proporcionada con Schema Registry se integra con Apache Flink. Para trabajar con Apache Flink, debe implementar las interfaces de [https://github.com/apache/flink/blob/master/flink-streaming-java/src/main/java/org/apache/flink/streaming/util/serialization/SerializationSchema.java](https://github.com/apache/flink/blob/master/flink-streaming-java/src/main/java/org/apache/flink/streaming/util/serialization/SerializationSchema.java) y [https://github.com/apache/flink/blob/8674b69964eae50cad024f2c5caf92a71bf21a09/flink-core/src/main/java/org/apache/flink/api/common/serialization/DeserializationSchema.java](https://github.com/apache/flink/blob/8674b69964eae50cad024f2c5caf92a71bf21a09/flink-core/src/main/java/org/apache/flink/api/common/serialization/DeserializationSchema.java), denominadas `GlueSchemaRegistryAvroSerializationSchema` y `GlueSchemaRegistryAvroDeserializationSchema`, que puede conectar a los conectores Apache Flink. ### Adición de una dependencia de AWS Glue Schema Registry en la aplicación Apache Flink Para configurar las dependencias de integración a AWS Glue Schema Registry en la aplicación Apache Flink: 1. Agregue la dependencia al archivo `pom.xml`. ``` software.amazon.glue schema-registry-flink-serde 1.0.0 ``` #### Integración de Kafka o Amazon MSK con Apache Flink Puede usar Managed Service para Apache Flink con Kafka como origen o receptor. **Kafka como origen** En el siguiente diagrama, se muestra la integración de Kinesis Data Streams con Managed Service para Apache Flink, con Kafka como origen. ![\[Kafka como origen.\]](http://docs.aws.amazon.com/es_es/glue/latest/dg/images/gsr-kafka-source.png) **Kafka como receptor** En el siguiente diagrama, se muestra la integración de Kinesis Data Streams con Managed Service para Apache Flink, con Kafka como receptor. ![\[Kafka como receptor.\]](http://docs.aws.amazon.com/es_es/glue/latest/dg/images/gsr-kafka-sink.png) Para integrar Kafka (o Amazon MSK) con Managed Service para Apache Flink, con Kafka como origen o receptor, realice los siguientes cambios de código. Agregue los bloques de código en negrita a su código respectivo en las secciones análogas. Si Kafka es el origen, entonces use el código deserializador (bloque 2). Si Kafka es el receptor, use el código serializador (bloque 3). ``` StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); String topic = "topic"; Properties properties = new Properties(); properties.setProperty("bootstrap.servers", "localhost:9092"); properties.setProperty("group.id", "test"); // block 1 Map 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 Puede usar Managed Service para Apache Flink, con Kinesis Data Streams como origen o como receptor. **Kinesis Data Streams como origen** En el siguiente diagrama, se muestra la integración de Kinesis Data Streams con Managed Service para Apache Flink, con Kinesis Data Streams como origen. ![\[Kinesis Data Streams como origen.\]](http://docs.aws.amazon.com/es_es/glue/latest/dg/images/gsr-kinesis-source.png) **Kinesis Data Streams como receptor** En el siguiente diagrama, se muestra la integración de Kinesis Data Streams con Managed Service para Apache Flink, con Kinesis Data Streams como receptor. ![\[Kinesis Data Streams como receptor.\]](http://docs.aws.amazon.com/es_es/glue/latest/dg/images/gsr-kinesis-sink.png) Para integrar Kinesis Data Streams con Managed Service para Apache Flink, con Kinesis Data Streams como origen o receptor, realice los cambios de código que se indican a continuación. Agregue los bloques de código en negrita a su código respectivo en las secciones análogas. Si Kinesis Data Streams es el origen, utilice el código deserializador (bloque 2). Si Kinesis Data Streams es el receptor, use el código serializador (bloque 3). ``` StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); String streamName = "stream"; Properties consumerConfig = new Properties(); consumerConfig.put(AWSConfigConstants.AWS_REGION, "aws-region"); consumerConfig.put(AWSConfigConstants.AWS_ACCESS_KEY_ID, "aws_access_key_id"); consumerConfig.put(AWSConfigConstants.AWS_SECRET_ACCESS_KEY, "aws_secret_access_key"); consumerConfig.put(ConsumerConfigConstants.STREAM_INITIAL_POSITION, "LATEST"); // block 1 Map 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 Para utilizar una función AWS Lambda como consumidor Apache Kafka/Amazon MSK y deserializar mensajes codificados por AVRO-con AWS Glue Schema Registry, visite la [página de MSK Labs](https://amazonmsk-labs.workshop.aws/en/msklambda/gsrschemareg.html). ## Caso de uso: AWS Glue Data Catalog Las tablas de AWS Glue soportan esquemas que se pueden especificar en forma manual o por referencia a AWS Glue Schema Registry. Schema Registry se integra con el Catálogo de datos para permitirle utilizar opcionalmente esquemas almacenados en Schema Registry al crear o actualizar tablas o particiones de AWS Glue en el Catálogo de datos. Para identificar una definición de esquema en Schema Registry, es necesario conocer, al menos, el ARN del esquema del que forma parte. Una versión de esquema, que contiene una definición de esquema, puede ser referenciada por su UUID o número de versión. Siempre hay una versión de esquema, la “última” versión, que se puede buscar sin saber su número de versión o UUID. Al llamar a las operaciones `CreateTable` o `UpdateTable`, transferirá una estructura `TableInput` que contiene un `StorageDescriptor`, que podría tener una `SchemaReference` a un esquema existente en Schema Registry. Del mismo modo, cuando se llama a las API `GetTable` o `GetPartition`, la respuesta puede contener el esquema y la `SchemaReference`. Cuando se crea una tabla o partición mediante referencias de esquema, el Catálogo de datos intentará buscar el esquema para esta referencia de esquema. En caso de que no pueda encontrar el esquema, Schema Registry devuelve un esquema vacío en la respuesta `GetTable`; de lo contrario, la respuesta tendrá el esquema y la referencia del esquema. Puede realizar las siguientes acciones desde la consola de AWS Glue. Para realizar estas operaciones y crear, actualizar o ver la información del esquema, debe brindar al usuario que realiza la llamada un rol de IAM que proporcione permisos para la API `GetSchemaVersion`. ### Agregar una tabla o actualizar el esquema de una tabla Agregar una nueva tabla a partir de un esquema existente enlaza la tabla a una versión de esquema específica. Una vez que se registren las nuevas versiones de esquema, puede actualizar esta definición de tabla desde la página View tables (Ver tabla) en la consola de AWS Glue o con la API [Acción UpdateTable (Python: update\$1table)](aws-glue-api-catalog-tables.md#aws-glue-api-catalog-tables-UpdateTable). #### Agregar una tabla a partir de un esquema existente Puede crear una tabla de AWS Glue a partir de una versión de esquema en el registro mediante la consola AWS Glue o la API `CreateTable`. **AWS GlueAPI** Al llamar a la API `CreateTable`, transferirá una `TableInput` que contiene un `StorageDescriptor` con una `SchemaReference` a un esquema existente en Schema Registry. **Consola de AWS Glue** Para crear una tabla desde la consola de AWS Glue: 1. Inicie sesión en Consola de administración de AWS y abra la consola de AWS Glue en [https://console.aws.amazon.com/glue/](https://console.aws.amazon.com/glue\). 1. En el panel de navegación, en **Data catalog** (Catálogo de datos), elija **Tables (Tablas)**. 1. En el menú **Add tables (Agregar tablas)**, elija **Add table from existing schema (Agregar tabla a partir del esquema existente)**. 1. Configure las propiedades de la tabla y el almacén de datos según la Guía para desarrolladores de AWS Glue. 1. En la página **Choose a Glue schema (Elegir un esquema de Glue)**, seleccione el **Record (Registro)** donde reside el esquema. 1. Elija el **Schema name (Nombre del esquema)** y seleccione la **Version (Versión)** del esquema que se va a aplicar. 1. Revise la previsualización del esquema y elija **Next (Siguiente)**. 1. Revise y cree la tabla. El esquema y la versión aplicados a la tabla aparecen en la columna **Glue schema (Esquema de Glue)** en la lista de tablas. Puede ver la tabla para ver más detalles. #### Actualización del esquema de una tabla Cuando esté disponible una nueva versión de esquema, es posible que desee actualizar el esquema de una tabla mediante la API [Acción UpdateTable (Python: update\$1table)](aws-glue-api-catalog-tables.md#aws-glue-api-catalog-tables-UpdateTable) o la consola de AWS Glue. **importante** Al actualizar el esquema de una tabla existente que tiene un esquema de AWS Glue especificado manualmente, el nuevo esquema al que se hace referencia en el Schema Registry puede ser incompatible. Esto puede dar lugar a que sus trabajos presenten errores. **AWS GlueAPI** Al llamar a la API `UpdateTable`, transferirá una `TableInput` que contiene un `StorageDescriptor` con una `SchemaReference` a un esquema existente en Schema Registry. **Consola de AWS Glue** Para actualizar el esquema de una tabla desde la consola de AWS Glue: 1. Inicie sesión en Consola de administración de AWS y abra la consola de AWS Glue en [https://console.aws.amazon.com/glue/](https://console.aws.amazon.com/glue\). 1. En el panel de navegación, en **Data catalog** (Catálogo de datos), elija **Tables (Tablas)**. 1. Vea la tabla de la lista de tablas. 1. Haga clic en **Update schema (Actualizar esquema)** en el cuadro que le informa sobre una nueva versión. 1. Revise las diferencias entre el esquema actual y el nuevo. 1. Seleccione **Show all schema differences (Mostrar todas las diferencias de esquemas)** para ver más detalles. 1. Seleccione **Save table (Guardar tabla)** para aceptar la nueva versión. ## Caso de uso: Streaming de AWS Glue AWS GlueEl streaming de consume datos de orígenes de streaming y realiza operaciones ETL antes de escribir en un receptor de salida. El origen del streaming de entrada se puede especificar mediante una tabla de datos o directamente especificando la configuración de origen. El streaming de AWS Glue admite una tabla del Catálogo de datos para el origen de transmisión creado con el esquema presente en AWS Glue Schema Registry. Puede crear un esquema en AWS Glue Schema Registry y, mediante el uso de este, crear una tabla de AWS Glue con un origen de streaming. Esta tabla de AWS Glue se puede utilizar como entrada para un trabajo de streaming de AWS Glue de manera de deserializar los datos en el flujo de entrada. Un punto que se debe tener en cuenta aquí es que, cuando cambia el esquema de AWS Glue Schema Registry, debe reiniciar el trabajo de streaming de AWS Glue para que el cambio se vea reflejado. ## Caso de uso: Apache Kafka Streams La API Apache Kafka Streams es una biblioteca cliente para procesar y analizar datos almacenados en Apache Kafka. Esta sección describe la integración de Apache Kafka Streams con AWS Glue Schema Registry, que le permite administrar y aplicar esquemas en sus aplicaciones de streaming de datos. Para obtener más información sobre Apache Kafka Streams, consulte [Apache Kafka Streams](https://kafka.apache.org/documentation/streams/). ### Integración con las bibliotecas SerDes Existe una clase de `GlueSchemaRegistryKafkaStreamsSerde` con la que puede configurar una aplicación de Streams. #### Código de ejemplo de aplicación de Kafka Streams Para utilizar el AWS Glue Schema Registry dentro de una aplicación Apache Kafka Streams: 1. Configure la aplicación Kafka Streams. ``` final Properties props = new Properties(); props.put(StreamsConfig.APPLICATION_ID_CONFIG, "avro-streams"); props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092"); props.put(StreamsConfig.CACHE_MAX_BYTES_BUFFERING_CONFIG, 0); props.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass().getName()); props.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, AWSKafkaAvroSerDe.class.getName()); props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest"); props.put(AWSSchemaRegistryConstants.AWS_REGION, "aws-region"); props.put(AWSSchemaRegistryConstants.SCHEMA_AUTO_REGISTRATION_SETTING, true); props.put(AWSSchemaRegistryConstants.AVRO_RECORD_TYPE, AvroRecordType.GENERIC_RECORD.getName()); props.put(AWSSchemaRegistryConstants.DATA_FORMAT, DataFormat.AVRO.name()); ``` 1. Cree un flujo a partir del tema avro-input. ``` StreamsBuilder builder = new StreamsBuilder(); final KStream 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 Estos resultados muestran el proceso de filtrado de registros que se filtraron en el paso 3 como color\$1favorito “rosa” o valor “15,0”. Registros antes del filtrado: ``` {"name": "Sansa", "favorite_number": 99, "favorite_color": "white"} {"name": "Harry", "favorite_number": 10, "favorite_color": "black"} {"name": "Hermione", "favorite_number": 1, "favorite_color": "red"} {"name": "Ron", "favorite_number": 0, "favorite_color": "pink"} {"name": "Jay", "favorite_number": 0, "favorite_color": "pink"} {"id": "commute_1","amount": 3.5} {"id": "grocery_1","amount": 25.5} {"id": "entertainment_1","amount": 19.2} {"id": "entertainment_2","amount": 105} {"id": "commute_1","amount": 15} ``` Registros después del filtrado: ``` {"name": "Sansa", "favorite_number": 99, "favorite_color": "white"} {"name": "Harry", "favorite_number": 10, "favorite_color": "black"} {"name": "Hermione", "favorite_number": 1, "favorite_color": "red"} {"name": "Ron", "favorite_number": 0, "favorite_color": "pink"} {"id": "commute_1","amount": 3.5} {"id": "grocery_1","amount": 25.5} {"id": "entertainment_1","amount": 19.2} {"id": "entertainment_2","amount": 105} ``` ### Caso de uso: Apache Kafka Connect La integración de Apache Kafka Connect con el AWS Glue Schema Registry permite obtener información de esquemas a partir de los conectores. Los convertidores Apache Kafka especifican el formato de datos dentro de Apache Kafka y cómo traducirlos a datos Apache Kafka Connect. Cada usuario de Apache Kafka Connect tendrá que configurar estos convertidores en función del formato en el que desea que sus datos estén cargados o almacenados en Apache Kafka. De esta manera, puede definir sus propios convertidores para traducir los datos de Apache Kafka Connect al tipo utilizado en AWS Glue Schema Registry (por ejemplo: Avro) y utilizar nuestro serializador para registrar su esquema y serializar. Los convertidores también pueden usar nuestro deserializador para deserializar los datos recibidos de Apache Kafka y volver a convertirlos en datos Apache Kafka Connect. A continuación se muestra un diagrama de flujo de trabajo de ejemplo. ![\[Flujo de trabajo Apache Kafka Connect.\]](http://docs.aws.amazon.com/es_es/glue/latest/dg/images/schema_reg_int_kafka_connect.png) 1. Instale el proyecto `aws-glue-schema-registry` al clonar el [repositorio de Github para AWS Glue Schema Registry](https://github.com/awslabs/aws-glue-schema-registry). ``` git clone git@github.com:awslabs/aws-glue-schema-registry.git cd aws-glue-schema-registry mvn clean install mvn dependency:copy-dependencies ``` 1. Si planea usar Apache Kafka Connect en modo *independiente*, actualice **connect-standalone.properties** según las instrucciones que se incluyen a continuación. Si planea usar Apache Kafka Connect en modo *distribuido*, actualice **connect-avro-distributed.properties** según las mismas instrucciones. 1. Agregue estas propiedades también al archivo de propiedades de conexión Apache Kafka: ``` key.converter.region=aws-region value.converter.region=aws-region key.converter.schemaAutoRegistrationEnabled=true value.converter.schemaAutoRegistrationEnabled=true key.converter.avroRecordType=GENERIC_RECORD value.converter.avroRecordType=GENERIC_RECORD ``` 1. Agregue el siguiente comando a la sección **Launch mode (Modo de lanzamiento)** en **kafka-run-class.sh**: ``` -cp $CLASSPATH:"/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). # Migración desde un registro de esquemas de terceros a AWS Glue Schema Registry La migración de un registro de esquema de terceros a AWS Glue Schema Registry tiene una dependencia en el registro de esquemas de terceros existente. Si hay registros en un tema de Apache Kafka que se enviaron mediante un registro de esquemas de terceros, los consumidores necesitan el registro de esquemas de terceros para deserializar esos registros. `AWSKafkaAvroDeserializer` proporciona la capacidad de especificar una clase de deserializador secundario que apunta al deserializador de terceros y se utiliza para deserializar esos registros. Existen dos criterios para retirar un esquema de terceros. En primer lugar, el retiro sólo puede ocurrir después de que los registros en temas de Apache Kafka que utilizan el registro de esquemas de terceros ya no sean requeridos por o para un consumidor. En segundo lugar, el retiro puede ocurrir si se eliminan los temas de Apache Kafka, dependiendo del período de retención especificado para esos temas. Tenga en cuenta que si tiene temas que tienen retención infinita, puede migrar a AWS Glue Schema Registry, pero no podrá retirar el registro de esquemas de terceros. Como solución alternativa, puede usar una aplicación o Mirror Maker 2 para leer el tema actual y producir un tema nuevo con AWS Glue Schema Registry. Migración desde un registro de esquemas de terceros a AWS Glue Schema Registry: 1. Cree un registro en AWS Glue Schema Registry o utilice el registro predeterminado. 1. Detenga el consumidor. Modifíquelo para incluir AWS Glue Schema Registry como el deserializador principal y el registro de esquemas de terceros como el secundario. + Configure las propiedades del consumidor. En este ejemplo, el secondary\$1deserializer (deserializador secundario) se configura en un deserializador diferente. El comportamiento es el siguiente: el consumidor recupera los registros de Amazon MSK y primero intenta utilizar el `AWSKafkaAvroDeserializer`. Si no puede leer el byte mágico que contiene el ID de esquema de Avro para AWS Glue Schema Registry, el `AWSKafkaAvroDeserializer` luego intenta usar la clase de deserializador proporcionada en el secondary\$1deserializer (deserializador secundario). Las propiedades específicas del deserializador secundario también deben proporcionarse en las propiedades del consumidor, como schema\$1registry\$1url\$1config (configuración url de registro de esquema) y specific\$1avro\$1reader\$1config (configuración de lector avro específica), como se muestra a continuación. ``` consumerProps.setProperty(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName()); consumerProps.setProperty(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, AWSKafkaAvroDeserializer.class.getName()); consumerProps.setProperty(AWSSchemaRegistryConstants.AWS_REGION, KafkaClickstreamConsumer.gsrRegion); consumerProps.setProperty(AWSSchemaRegistryConstants.SECONDARY_DESERIALIZER, KafkaAvroDeserializer.class.getName()); consumerProps.setProperty(KafkaAvroDeserializerConfig.SCHEMA_REGISTRY_URL_CONFIG, "URL for third-party schema registry"); consumerProps.setProperty(KafkaAvroDeserializerConfig.SPECIFIC_AVRO_READER_CONFIG, "true"); ``` 1. Reinicie el consumidor. 1. Detenga el productor y dirija al productor a AWS Glue Schema Registry. 1. Configure las propiedades del productor. En este ejemplo, el productor utilizará las versiones de esquema de registro predeterminado y registro automático. ``` producerProps.setProperty(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName()); producerProps.setProperty(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, AWSKafkaAvroSerializer.class.getName()); producerProps.setProperty(AWSSchemaRegistryConstants.AWS_REGION, "us-east-2"); producerProps.setProperty(AWSSchemaRegistryConstants.AVRO_RECORD_TYPE, AvroRecordType.SPECIFIC_RECORD.getName()); producerProps.setProperty(AWSSchemaRegistryConstants.SCHEMA_AUTO_REGISTRATION_SETTING, "true"); ``` 1. (Opcional) mueva manualmente los esquemas y las versiones de esquema existentes del registro de esquemas de terceros actual a AWS Glue Schema Registry, ya sea al registro predeterminado en AWS Glue Schema Registry o a un registro no predeterminado específico en AWS Glue Schema Registry. Esto se puede hacer al exportar esquemas de los registros de esquemas de terceros en formato JSON y crear nuevos esquemas en AWS Glue Schema Registry con Consola de administración de AWS o AWS CLI. Este paso puede ser importante si necesita habilitar las comprobaciones de compatibilidad con versiones de esquema anteriores para las versiones de esquema recién creadas mediante AWS CLI y Consola de administración de AWS, o cuando los productores envían mensajes con un nuevo esquema con la función de registro automático de versiones de esquema activada. 1. Inicie el productor.