Las traducciones son generadas a través de traducción automática. En caso de conflicto entre la traducción y la version original de inglés, prevalecerá la version en inglés.
# Datos de escritura en Amazon Kinesis Data Streams
Un *productor* es una aplicación que escribe datos en Amazon Kinesis Data Streams. Puede crear productores para Kinesis Data Streams mediante la Biblioteca de productores de Kinesis (KPL) y AWS SDK para Java la Biblioteca de productores de Kinesis.
Si es la primera vez que utiliza Kinesis Data Streams, familiarícese antes con los conceptos y los términos que encontrará en [¿Qué es Amazon Kinesis Data Streams?](introduction.md) y [Úselo AWS CLI para realizar operaciones de Amazon Kinesis Data Streams](getting-started.md).
**importante**
Kinesis Data Streams admite cambios en el periodo de retención de los registros de datos del flujo de datos. Para obtener más información, consulte [Cambiar el periodo de retención de datos](kinesis-extended-retention.md).
Para insertar datos en la secuencia, debe especificar el nombre de la secuencia, una clave de fragmento y el blob de datos que se han de agregar a la secuencia. La clave de partición se utiliza para determinar en qué fragmento de la secuencia se agregará el registro de datos.
Todos los datos del fragmento se envían al mismo proceso de trabajo que procesa el fragmento. La clave de partición que utilice dependerá de la lógica de su aplicación. El número de claves de partición normalmente debe ser mucho mayor que el número de fragmentos. Esto se debe a que la clave de partición se utiliza para determinar cómo asignar un registro de datos a un fragmento determinado. Si tiene suficientes claves de partición, los datos pueden distribuirse con uniformidad entre todos los fragmentos de una secuencia.
**Topics**
+ [Desarrollo de productores con Amazon Kinesis Producer Library (KPL)](developing-producers-with-kpl.md)
+ [Desarrolle productores mediante la API Amazon Kinesis Data Streams con AWS SDK para Java](developing-producers-with-sdk.md)
+ [Escrivir en Amazon Kinesis Data Streams mediante el agente de Kinesis](writing-with-agents.md)
+ [Escriba en Kinesis Data Streams mediante otros servicios AWS](using-other-services.md)
+ [Escribir en Kinesis Data Streams mediante integraciones de terceros](using-other-services-third-party.md)
+ [Solucionar problemas de los productores de Amazon Kinesis Data Streams](troubleshooting-producers.md)
+ [Optimizar sus productores de Amazon Kinesis Data Streams](advanced-producers.md)
# Desarrollo de productores con Amazon Kinesis Producer Library (KPL)
Un productor de Amazon Kinesis Data Streams es una aplicación que coloca los registros de datos de usuario en un flujo de datos de Kinesis Data Streams (también denominado *ingesta de datos*). Amazon Kinesis Producer Library (KPL) simplifica el desarrollo de aplicaciones productoras y permite a los desarrolladores alcanzar un mayor rendimiento de escritura en un flujo de datos de Kinesis Data Streams.
Puedes monitorizar la KPL con Amazon CloudWatch. Para obtener más información, consulte [Supervise la biblioteca de Kinesis Producer con Amazon CloudWatch](monitoring-with-kpl.md).
**Topics**
+ [Revisar el rol de KPL](#developing-producers-with-kpl-role)
+ [Descubrir las ventajas del uso de KPL](#developing-producers-with-kpl-advantage)
+ [Comprender cuándo no usar la KPL](#developing-producers-with-kpl-when)
+ [Instalar el KPL](kinesis-kpl-dl-install.md)
+ [Migrar de KPL 0.x a KPL 1.x](kpl-migration-1x.md)
+ [Transicionar a los certificados de Amazon Trust Services (ATS) para la KPL](kinesis-kpl-upgrades.md)
+ [Plataformas compatibles con KPL](kinesis-kpl-supported-plats.md)
+ [Conceptos clave de KPL](kinesis-kpl-concepts.md)
+ [Integrar la KPL con el código de productor](kinesis-kpl-integration.md)
+ [Escribir en Kinesis Data Stream mediante KPL](kinesis-kpl-writing.md)
+ [Configurar Amazon Kinesis Producer Library](kinesis-kpl-config.md)
+ [Implementar la desagregación de consumidores](kinesis-kpl-consumer-deaggregation.md)
+ [Usar KPL con Amazon Data Firehose](kpl-with-firehose.md)
+ [Utilice el KPL con el registro de AWS Glue esquemas](kpl-with-schemaregistry.md)
+ [Configurar la configuración del proxy de KPL](kpl-proxy-configuration.md)
+ [Política de ciclo de vida de la versión de KPL](kpl-version-lifecycle-policy.md)
**nota**
Se recomienda actualizar a la versión más reciente de KPL. KPL se actualiza periódicamente con versiones más recientes que incluyen los últimos parches de dependencia y seguridad, correcciones de errores y nuevas características compatibles con versiones anteriores. Para obtener más información, consulte [https://github.com/awslabs/amazon-kinesis-producer/releases/](https://github.com/awslabs/amazon-kinesis-producer/releases/).
## Revisar el rol de KPL
La KPL es una easy-to-use biblioteca altamente configurable que le ayuda a escribir en una transmisión de datos de Kinesis. Ejerce de intermediaria entre el código de la aplicación de productor y las acciones de la API de Kinesis Data Streams. KPL realiza las siguientes tareas principales:
+ Escribe en uno o varios flujo de datos de Kinesis con un mecanismo de reintento automático y configurable
+ Recopila registros y utiliza `PutRecords` para escribir varios registros en varios fragmentos por solicitud
+ Acumula registros de usuario para aumentar el tamaño de la carga y mejorar el rendimiento
+ Se integra perfectamente con [Kinesis Client Library](https://docs.aws.amazon.com/kinesis/latest/dev/developing-consumers-with-kcl.html) (KCL) para desagrupar registros en lotes en el consumidor
+ Envía las CloudWatch estadísticas de Amazon en tu nombre para proporcionar visibilidad sobre el rendimiento del productor
Tenga en cuenta que la KPL es diferente de la API de Kinesis Data Streams que está disponible en. [AWS SDKs](https://aws.amazon.com/tools/) La API de Kinesis Data Streams ayuda a administrar muchos aspectos de Kinesis Data Streams (como la creación de flujos, el cambio de las particiones de registros, así como la inserción y obtención de estos), mientras que KPL proporciona una capa de abstracción específica para la ingesta de datos. Para obtener información sobre la API de Kinesis Data Streams, consulte [Referencia de la API de Amazon Kinesis](https://docs.aws.amazon.com/kinesis/latest/APIReference/).
## Descubrir las ventajas del uso de KPL
La siguiente lista representa algunas de las ventajas más importantes de utilizar KPL para el desarrollo de productores de Kinesis Data Streams.
KPL puede utilizarse en casos de uso síncronos o asíncronos. Se recomienda utilizar el mayor rendimiento de la interfaz asíncrona a menos que haya una razón específica para utilizar un comportamiento síncrono. Para obtener más información sobre estos dos casos de uso y ver código de muestra, consulte [Escribir en Kinesis Data Stream mediante KPL](kinesis-kpl-writing.md).
**Beneficios de rendimiento**
KPL puede ayudar a crear productores de alto rendimiento. Piense en una situación en la que las instancias de Amazon EC2 sirven como un proxy para recopilar eventos de 100 bytes de cientos o miles de dispositivos de baja potencia y para escribir registros en un flujo de datos de Kinesis. Cada una de estas instancias EC2 debe escribir miles de eventos por segundo en la secuencia de datos. Para conseguir el nivel de rendimiento necesario, los productores deben implementar lógica complicada como, por ejemplo, la creación de lotes o la ejecución de subprocesos, además de lógica de reintento y desagrupación de registros en el lado del consumidor. KPL realiza todas estas tareas.
**Facilidad de uso en el lado del consumidor**
En el caso de los desarrolladores del lado del consumidor que usan KCL en Java, KPL se integra sin mayor esfuerzo. Cuando KCL recupera un registro agregado de Kinesis Data Streams que consta de varios registros de usuario de KPL, llama automáticamente a KPL para que extraiga los registro de usuario individuales antes de devolverlos al usuario.
Para los desarrolladores del lado del consumidor que no utilicen KCL, sino que, en su lugar, utilicen la operación de la API `GetRecords` directamente, hay disponible una biblioteca de Java de KPL que pueden usar para extraer los registros de los usuarios individuales antes de devolverlos al usuario.
**Monitoreo del productor**
Puede recopilar, supervisar y analizar a sus productores de Kinesis Data Streams mediante CloudWatch Amazon y la KPL. El KPL emite el rendimiento, los errores y otras métricas CloudWatch en su nombre, y se puede configurar para supervisarlo a nivel de transmisión, fragmento o productor.
**Arquitectura asíncrona**
Debido a que KPL puede almacenar en búfer los registros antes de enviarlos a Kinesis Data Streams, en la aplicación intermediaria no se fuerza el bloqueo y la espera a la confirmación de que el registro ha llegado al servidor antes de continuar con el tiempo de ejecución. Una llamada para insertar un registro en KPL siempre vuelve inmediatamente, y no se espera a que el registro se envíe o a que se reciba una respuesta desde el servidor. En su lugar, se crea un objeto `Future` que recibe el resultado de enviar el registro a Kinesis Data Streams en un momento posterior. Este comportamiento es el mismo que el de los clientes asíncronos del SDK. AWS
## Comprender cuándo no usar la KPL
KPL puede ocasionar un retraso adicional en el procesamiento de hasta `RecordMaxBufferedTime` en la biblioteca (configurable por el usuario). Los valores más altos de `RecordMaxBufferedTime` ofrecen un mayor eficacia de empaquetamiento y un mejor rendimiento. Es posible que las aplicaciones que no pueden tolerar este retraso tengan que utilizar el SDK de AWS directamente. Para obtener más información sobre el uso del AWS SDK con Kinesis Data Streams, [Desarrolle productores mediante la API Amazon Kinesis Data Streams con AWS SDK para Java](developing-producers-with-sdk.md) consulte. Para más información sobre `RecordMaxBufferedTime` y otras propiedades configurables por el usuario de KPL, consulte [Configurar Amazon Kinesis Producer Library](kinesis-kpl-config.md).
# Instalar el KPL
Amazon proporciona archivos binarios precompilados de Amazon Kinesis Producer Library (KPL) en C\$1\$1 para macOS, Windows y distribuciones recientes de Linux (para información detallada sobre las plataformas compatibles, consulte la siguiente sección). Estos archivos binarios se empaquetan como parte de archivos .jar de Java y se invocan y utilizan automáticamente si se usa Maven para instalar el paquete. Para encontrar las versiones más recientes de KPL y KCL, utilice los siguientes enlaces de búsqueda de Maven:
+ [KPL](https://search.maven.org/#search|ga|1|amazon-kinesis-producer)
+ [KCL](https://search.maven.org/#search|ga|1|amazon-kinesis-client)
Los archivos binarios de Linux han sido compilados con la recopilación del compilador GNU (GCC) y están vinculados estáticamente con libstdc\$1\$1 en Linux. Lo esperable es que funcionen en cualquier distribución de Linux de 64 bits que incluya una versión de glibc 2.5 o superior.
Los usuarios de distribuciones Linux anteriores pueden compilar el KPL siguiendo las instrucciones de compilación incluidas en el código fuente. GitHub Para descargar el KPL GitHub, consulte la biblioteca [Amazon Kinesis](https://github.com/awslabs/amazon-kinesis-producer) Producer Library.
**importante**
Amazon Kinesis Producer Library (KPL) 0.x estará disponible el 30 de end-of-support enero de 2026. **Recomendamos** migrar las aplicaciones de KCL que utilizan la versión 0.x a la versión más reciente de KPL antes del 30 de enero de 2026. Para encontrar la versión más reciente de KPL, consulte la [página de KPL en Github](https://github.com/awslabs/amazon-kinesis-producer). Para obtener información sobre cómo migrar de KCL 0.x a KCL 1.x, consulte [Migrar de KPL 0.x a KPL 1.x](kpl-migration-1x.md).
# Migrar de KPL 0.x a KPL 1.x
En este tema se proporcionan step-by-step instrucciones para migrar a su consumidor de KPL 0.x a KPL 1.x. KPL 1.x introduce la compatibilidad con la versión AWS SDK para Java 2.x y, al mismo tiempo, mantiene la compatibilidad de la interfaz con las versiones anteriores. No es necesario actualizar su lógica principal de procesamiento de datos para migrar a KPL 1.x.
1. **Asegúrese de que tenga los siguientes requisitos previos:**
+ Java Development Kit (JDK) 8 o posterior,
+ AWS SDK para Java 2.x
+ Maven o Gradle para la administración de dependencias.
1. **Añadir dependencias**
Si está utilizando Maven, agregue la siguiente dependencia a su archivo pom.xml. Asegúrese de actualizar groupId de `com.amazonaws` a `software.amazon.kinesis` y la versión `1.x.x` a la versión más reciente de KPL.
```
software.amazon.kinesis
amazon-kinesis-producer
1.x.x
```
Si está utilizando Gradle, agregue lo siguiente a su archivo `build.gradle`. Asegúrese de reemplazar `1.x.x` por la versión más reciente de KPL.
```
implementation 'software.amazon.kinesis:amazon-kinesis-producer:1.x.x'
```
Puede buscar la versión más reciente del KPL en el [Repositorio central de Maven](https://central.sonatype.com/search?q=amazon-kinesis-producer).
1. **Actualizar las declaraciones de importación de KPL**
KPL 1.x usa la versión AWS SDK para Java 2.x y usa un nombre de paquete actualizado que comienza por`software.amazon.kinesis`, en comparación con el nombre de paquete del KPL anterior que comenzaba por. `com.amazonaws.services.kinesis`
Sustituya la importación de `com.amazonaws.services.kinesis` con `software.amazon.kinesis`. En la siguiente tabla se muestran las importaciones que se deben reemplazar.
**Importar reemplazos**
[\[See the AWS documentation website for more details\]](http://docs.aws.amazon.com/es_es/streams/latest/dev/kpl-migration-1x.html)
1. **Actualice las declaraciones de importación para AWS las clases de proveedores de credenciales**
Al migrar a KPL 1.x, debe actualizar los paquetes y las clases que importe en el código de la aplicación KPL que estén basados en la versión AWS SDK para Java 1.x por los correspondientes basados en la versión 2.x. AWS SDK para Java Las importaciones habituales en la aplicación KPL son las clases de proveedores de credenciales. Consulte los [cambios en el proveedor de credenciales](https://docs.aws.amazon.com/sdk-for-java/latest/developer-guide/migration-client-credentials.html) en la documentación de la guía de migración de la AWS SDK para Java versión 2.x para ver la lista completa de cambios en el proveedor de credenciales. Este es el cambio de importación habitual que necesitará para realizar en sus aplicaciones de KPL.
**Importar en KPL 0.x**
```
import com.amazonaws.auth.DefaultAWSCredentialsProviderChain;
```
**Importar en KPL 1.x**
```
import software.amazon.awssdk.auth.credentials.DefaultCredentialsProvider;
```
Si importa otros proveedores de credenciales basados en la versión AWS SDK para Java 1.x, debe actualizarlos por los equivalentes a la versión AWS SDK para Java 2.x. Si no has importado ninguno classes/packages de la versión AWS SDK para Java 1.x, puedes ignorar este paso.
1. **Actualice la configuración del proveedor de credenciales en la configuración de KPL**
La configuración del proveedor de credenciales en KPL 1.x requiere los proveedores de credenciales AWS SDK para Java 2.x. Si va a pasar los proveedores de credenciales de la versión AWS SDK para Java 1.x a la `KinesisProducerConfiguration` sustituyendo el proveedor de credenciales predeterminado, debe actualizarlo con los proveedores de credenciales de la versión AWS SDK para Java 2.x. Consulte [los cambios en el proveedor de credenciales](https://docs.aws.amazon.com/sdk-for-java/latest/developer-guide/migration-client-credentials.html) en la documentación de la guía de migración de la AWS SDK para Java versión 2.x para ver la lista completa de cambios en el proveedor de credenciales. Si no anuló el proveedor de credenciales predeterminado en la configuración de KPL, puede omitir este paso.
Por ejemplo, si va a anular el proveedor de credenciales predeterminado para KPL con el siguiente código:
```
KinesisProducerConfiguration config = new KinesisProducerConfiguration();
// SDK v1 default credentials provider
config.setCredentialsProvider(new DefaultAWSCredentialsProviderChain());
```
Debe actualizarlos con el siguiente código para usar el proveedor de credenciales AWS SDK para Java 2.x:
```
KinesisProducerConfiguration config = new KinesisProducerConfiguration();
// New SDK v2 default credentials provider
config.setCredentialsProvider(DefaultCredentialsProvider.create());
```
# Transicionar a los certificados de Amazon Trust Services (ATS) para la KPL
El 9 de febrero de 2018, a las 9:00 h (PST), Amazon Kinesis Data Streams instaló certificados ATS. Si desea continuar escribiendo registros en Kinesis Data Streams con Amazon Kinesis Producer Library (KPL) debe actualizar la instalación de KPL a la [versión 0.12.6](http://search.maven.org/#artifactdetails|com.amazonaws|amazon-kinesis-producer|0.12.6|jar) o posterior. Este cambio afecta a todas las AWS regiones.
Para obtener información sobre el cambio a ATS, consulte [Cómo prepararse para AWS el traslado a su propia autoridad de certificación](https://aws.amazon.com/blogs/security/how-to-prepare-for-aws-move-to-its-own-certificate-authority/).
Si tiene problemas y necesita soporte técnico, [cree un caso](https://console.aws.amazon.com/support/v1#/case/create) en el Centro de AWS Support.
# Plataformas compatibles con KPL
Amazon Kinesis Producer Library (KPL) está escrito en C\$1\$1 y se ejecuta como un proceso secundario del proceso principal de usuario. Los binarios nativos precompilados de 64 bits se empaquetan con la versión de Java y vuelven a ser administrados por la clase contenedora Java.
El paquete de Java se ejecuta sin necesidad de instalar otras bibliotecas en los siguientes sistemas operativos:
+ Distribuciones de Linux con kernel 2.6.18 (septiembre de 2006) y posteriores
+ Apple iOS X 10.9 y versiones posteriores
+ Windows Server 2008 y posteriores
**importante**
Windows Server 2008 y las versiones posteriores son compatibles con todas las versiones de KPL hasta la versión 0.14.0.
La plataforma Windows NO es compatible a partir de la versión 0.14.0 o superior de KPL.
Tenga en cuenta que KPL solo es compatible con sistemas de 64 bits.
## Código fuente
Si los archivos binarios proporcionados en la instalación de KPL no son suficientes para el entorno, el núcleo de KPL se escribirá como un módulo C\$1\$1. El código fuente del módulo C\$1\$1 y la interfaz Java se publican bajo la licencia pública de Amazon y están disponibles GitHub en la biblioteca de [productores de Amazon Kinesis](https://github.com/awslabs/amazon-kinesis-producer). Aunque KPL se puede utilizar en cualquier plataforma en la que estén disponibles un compilador C\$1\$1 y un JRE que cumplan los estándares recientes, Amazon no admite oficialmente ninguna plataforma que no figure en la lista de plataformas compatibles.
# Conceptos clave de KPL
Las siguientes secciones contienen conceptos y términos necesarios para comprender y utilizar Amazon Kinesis Producer Library (KPL).
**Topics**
+ [Registros](#kinesis-kpl-concepts-records)
+ [Agrupación en lotes](#kinesis-kpl-concepts-batching)
+ [Agregación](#kinesis-kpl-concepts-aggretation)
+ [Recopilación](#kinesis-kpl-concepts-collection)
## Registros
En esta guía, distinguimos entre *registros de usuario de KPL* y *registros de Kinesis Data Streams*. Cuando utilizamos el término *registro* sin ningún calificador, nos referimos a un *registro de usuario de KPL*. Cuando nos referimos a un registro de Kinesis Data Streams, decimos explícitamente *registro de Kinesis Data Streams*.
Un registro de usuario de KPL es un blob de datos que tiene un significado determinado para el usuario. Entre los ejemplos se incluyen un blob JSON que representa un evento de la interfaz de usuario en un sitio web, o una entrada de registro de un servidor web.
Un registro de Kinesis Data Streams es una instancia de la estructura de datos de `Record` definida por la API del servicio Kinesis Data Streams. Contiene una clave de partición, un número secuencial y un blob de datos.
## Agrupación en lotes
La *agrupación en lotes* se refiere a la realización de una sola acción en varios elementos en lugar de realizar la acción repetidamente en cada elemento individual.
En este contexto, el “elemento” es un registro, y la acción es enviarlo a Kinesis Data Streams. En una situación sin agrupación en lotes, cada registro se insertaría en un registro de Kinesis Data Streams independiente y se hará una solicitud HTTP para enviarlo a Kinesis Data Streams. Al agrupar en lotes, cada solicitud HTTP puede tener varios registros en lugar de solo uno.
KPL admite dos tipos de agrupación en lotes:
+ *Agregación*: almacenamiento de varios registros en un único registro de Kinesis Data Streams.
+ *Recopilación*: uso de la operación de la API `PutRecords` para enviar varios registros de Kinesis Data Streams a una o varias particiones del flujo de datos de Kinesis.
Los dos tipos de agrupación en lote de KPL se diseñaron para coexistir y se pueden activar o desactivar de forma independiente entre sí. Ambos están activados de forma predeterminada.
## Agregación
La *agregación* se refiere al almacenamiento de varios registros en un registro de Kinesis Data Streams. La agregación permite a los clientes aumentar el número de registros enviados por cada llamada a la API, lo que aumenta el rendimiento del productor de manera eficaz.
Las particiones de Kinesis Data Streams admiten hasta 1000 registros de Kinesis Data Streams por segundo o un rendimiento de 1 MB. El límite de registros de Kinesis Data Streams por segundo fuerza a los consumidores a usar registros de tamaño inferior a 1 KB. La agregación de registros en lote permite a los clientes combinar varios registros en un único registro de Kinesis Data Streams. Esto permite a los clientes mejorar su rendimiento por fragmento.
Piense en el caso de una partición en la región us-east-1 que se está ejecutando a una velocidad constante de 1000 registros por segundo, con registros de 512 bytes cada uno. Con la agregación de KPL puede empaquetar 1000 registros en solo 10 registros de Kinesis Data Streams, de manera que el número de registros por segundo se reduce a 10 (con 50 KB cada uno).
## Recopilación
La *recopilación* se refiere a agrupar en lote varios registros de Kinesis Data Streams y enviarlos en una única solicitud HTTP con una llamada a la operación de la API `PutRecords`, en lugar de enviar cada registro de Kinesis Data Streams en su propia solicitud HTTP.
Esto aumenta el rendimiento, si lo comparamos con una situación en la que no se use recopilación, ya que reduce la sobrecarga que implica realizar muchas solicitudes de HTTP independientes. De hecho, `PutRecords` en sí se ha diseñado específicamente para este fin.
La recopilación es diferente de la agregación, ya que funciona con grupos de registros de Kinesis Data Streams. Los registros de Kinesis Data Streams que se recopilan pueden contener varios registros del usuario. La relación se puede visualizar así:
```
record 0 --|
record 1 | [ Aggregation ]
... |--> Amazon Kinesis record 0 --|
... | |
record A --| |
|
... ... |
|
record K --| |
record L | | [ Collection ]
... |--> Amazon Kinesis record C --|--> PutRecords Request
... | |
record S --| |
|
... ... |
|
record AA--| |
record BB | |
... |--> Amazon Kinesis record M --|
... |
record ZZ--|
```
# Integrar la KPL con el código de productor
Amazon Kinesis Producer Library (KPL) se ejecuta en un proceso independiente y se comunica con el proceso principal de usuario mediante IPC. Esta arquitectura se denomina a veces [microservicio](http://en.wikipedia.org/wiki/Microservices), y se elige por dos razones principales:
**1) El proceso de usuario no se bloqueará aunque KPL se bloquee**
El proceso podría tener tareas no relacionadas con Kinesis Data Streams y así podría continuar la operación incluso si KPL se bloquea. También es posible que el proceso principal de usuario reinicie KPL y se recupere hasta alcanzar un estado totalmente funcional (esta funcionalidad se encuentra en los contenedores oficiales).
Un ejemplo es un servidor web que envía métricas a Kinesis Data Streams. El servidor puede continuar sirviendo páginas incluso si la parte de Kinesis Data Streams dejó de funcionar. El bloqueo de todo el servidor como consecuencia de un error en KPL provocaría una interrupción innecesaria del servicio.
**2) Compatibilidad con clientes arbitrarios**
Siempre hay consumidores que usan lenguajes diferentes a los que son oficialmente compatibles. Estos usuarios también deberían poder utilizar KPL fácilmente.
## Matriz de uso recomendada
La siguiente matriz de uso enumera la configuración recomendada para diferentes usuarios y le recomienda si utilizar KPL y cómo hacerlo. Tenga en cuenta que si se habilita la agregación, también debe utilizar desagrupación para extraer sus registros en el lado del consumidor.
| Lenguaje del lado del productor | Lenguaje del lado del consumidor | Versión de KCL | Lógica del punto de comprobación | ¿Puede utilizar KPL? | Advertencias |
| --- | --- | --- | --- | --- | --- |
| Cualquiera excepto Java | \$1 | \$1 | \$1 | No | N/A |
| Java | Java | Utiliza el SDK de Java directamente | N/A | Sí | Si se utiliza agregación, deberá usar la biblioteca de desagrupación facilitada tras la llamada de GetRecords. |
| Java | Cualquiera excepto Java | Utiliza SDK directamente | N/A | Sí | Debe deshabilitar la agregación. |
| Java | Java | 1.3.x | N/A | Sí | Debe deshabilitar la agregación. |
| Java | Java | 1.4.x | Llama al punto de comprobación sin argumentos | Sí | Ninguno |
| Java | Java | 1.4.x | Llama al punto de comprobación con un número secuencial explícito | Sí | Desactive la agregación o cambie el código para usar números secuenciales ampliados para la creación de puntos de control. |
| Java | Cualquiera excepto Java | 1.3.x \$1 daemon multilenguaje \$1 contenedor específico del lenguaje | N/A | Sí | Debe deshabilitar la agregación. |
# Escribir en Kinesis Data Stream mediante KPL
En las secciones siguientes, se muestra un código de ejemplo con una progresión que va desde el productor más básico hasta un código completamente asíncrono.
## Código de productor básico
El siguiente código es todo lo que necesita para escribir un productor mínimamente funcional. Los registros de usuario de Amazon Kinesis Producer Library (KPL) se procesan en segundo plano.
```
// KinesisProducer gets credentials automatically like
// DefaultAWSCredentialsProviderChain.
// It also gets region automatically from the EC2 metadata service.
KinesisProducer kinesis = new KinesisProducer();
// Put some records
for (int i = 0; i < 100; ++i) {
ByteBuffer data = ByteBuffer.wrap("myData".getBytes("UTF-8"));
// doesn't block
kinesis.addUserRecord("myStream", "myPartitionKey", data);
}
// Do other stuff ...
```
## Responder a los resultados de manera síncrona
En el ejemplo anterior, el código no comprobó si el procesamiento de registros de usuario de KPL finalizó correctamente. KPL efectúa los reintentos necesarios en caso de error. Sin embargo, si desea comprobar los resultados, puede examinarlos con los objetos `Future` que devuelve `addUserRecord`, como en el siguiente ejemplo (el ejemplo anterior se incluye para aportar contexto):
```
KinesisProducer kinesis = new KinesisProducer();
// Put some records and save the Futures
List> putFutures = new LinkedList>();
for (int i = 0; i < 100; i++) {
ByteBuffer data = ByteBuffer.wrap("myData".getBytes("UTF-8"));
// doesn't block
putFutures.add(
kinesis.addUserRecord("myStream", "myPartitionKey", data));
}
// Wait for puts to finish and check the results
for (Future f : putFutures) {
UserRecordResult result = f.get(); // this does block
if (result.isSuccessful()) {
System.out.println("Put record into shard " +
result.getShardId());
} else {
for (Attempt attempt : result.getAttempts()) {
// Analyze and respond to the failure
}
}
}
```
## Responder a los resultados de manera asíncrona
El ejemplo anterior llama a `get()` en un objeto `Future` que bloquea el tiempo de ejecución. Si no desea bloquear el tiempo de ejecución, puede utilizar una devolución de llamada asíncrona, tal y como se muestra en el ejemplo siguiente:
```
KinesisProducer kinesis = new KinesisProducer();
FutureCallback myCallback = new FutureCallback() {
@Override public void onFailure(Throwable t) {
/* Analyze and respond to the failure */
};
@Override public void onSuccess(UserRecordResult result) {
/* Respond to the success */
};
};
for (int i = 0; i < 100; ++i) {
ByteBuffer data = ByteBuffer.wrap("myData".getBytes("UTF-8"));
ListenableFuture f = kinesis.addUserRecord("myStream", "myPartitionKey", data);
// If the Future is complete by the time we call addCallback, the callback will be invoked immediately.
Futures.addCallback(f, myCallback);
}
```
# Configurar Amazon Kinesis Producer Library
Aunque la configuración predeterminada debería funcionar sin problemas para la mayoría de los casos de uso, es posible que desee cambiar algunos de los ajustes predeterminados para adaptar el comportamiento de `KinesisProducer` a sus necesidades. Para hacerlo, se puede pasar una instancia de la clase `KinesisProducerConfiguration` al constructor `KinesisProducer`, por ejemplo:
```
KinesisProducerConfiguration config = new KinesisProducerConfiguration()
.setRecordMaxBufferedTime(3000)
.setMaxConnections(1)
.setRequestTimeout(60000)
.setRegion("us-west-1");
final KinesisProducer kinesisProducer = new KinesisProducer(config);
```
También puede cargar una configuración desde un archivo de propiedades:
```
KinesisProducerConfiguration config = KinesisProducerConfiguration.fromPropertiesFile("default_config.properties");
```
Puede sustituir la ruta y el nombre de archivo a los que tiene acceso el proceso del usuario. También puede llamar a métodos fijos en la instancia `KinesisProducerConfiguration` que hayan sido creados así para personalizar la configuración.
El archivo de propiedades debe especificar los parámetros con sus nombres en PascalCase. Los nombres han de coincidir con los utilizados en los métodos fijos de la clase `KinesisProducerConfiguration`. Por ejemplo:
```
RecordMaxBufferedTime = 100
MaxConnections = 4
RequestTimeout = 6000
Region = us-west-1
```
Para obtener más información sobre las reglas de uso de los parámetros de configuración y los límites de valores, consulte el [ejemplo del archivo de propiedades de configuración en GitHub](https://github.com/awslabs/amazon-kinesis-producer/blob/master/java/amazon-kinesis-producer-sample/default_config.properties).
Tenga en cuenta que después de inicializar `KinesisProducer`, los cambios en la instancia `KinesisProducerConfiguration` no tienen efecto alguno. Actualmente, `KinesisProducer` no admite la reconfiguración dinámica.
# Implementar la desagregación de consumidores
A partir de la versión 1.4.0, KCL es compatible con la desagrupación automática de registros de usuario de KPL. El código de la aplicación de consumidor escrito con versiones anteriores de KCL se compilará sin ningún tipo de modificación después de actualizar KCL. Sin embargo, si la agregación de KPL se utiliza en el lado del productor, existe un matiz relacionado con la creación de puntos de control: todos los subregistros de un registro agregado tienen el mismo número de secuencia, de modo que los datos adicionales deben almacenarse con el punto de control si es necesario hacer una distinción entre subregistros. Estos datos adicionales se denominan *números subsecuenciales*.
**Topics**
+ [Migrar de versiomrd anteriores de KCL](#kinesis-kpl-consumer-deaggregation-migration)
+ [Usar extensiones de KCL para la desagrupación de KPL](#kinesis-kpl-consumer-deaggregation-extensions)
+ [Úselo GetRecords directamente](#kinesis-kpl-consumer-deaggregation-getrecords)
## Migrar de versiomrd anteriores de KCL
No es necesario cambiar las llamadas existentes para crear puntos de control con la agregación. La recuperación de todos los registros almacenados correctamente en Kinesis Data Streams seguirá estando garantizada. Ahora, KCL ofrece dos nuevas operaciones de puntos de control para casos de uso particulares que se describen a continuación.
Si el código existente se hubiera escrito para KCL antes de que fuera compatible con KPL y se llama a la operación del punto de control sin argumentos, esto equivaldría a la creación de puntos de control para el número de secuencia del último registro de usuario de KPL en el lote. Si la operación de punto de comprobación se llama con una cadena de número secuencial, equivaldrá a la creación de un punto de comprobación para el número secuencial determinado del lote junto con el número subsecuencial implícito 0 (cero).
La llamada a la nueva operación de punto de control de KCL `checkpoint()` sin argumentos equivale semánticamente a hacer un punto de control del número de secuencia de la última llamada de `Record` en el lote, junto con el número subsecuencial implícito 0 (cero).
La llamada a la nueva operación de punto de control de KCL `checkpoint(Record record)` equivale semánticamente a hacer un punto de control del número secuencial del `Record` dado, junto con el número subsecuencial implícito 0 (cero). Si la llamada al `Record` es realmente un `UserRecord`, el número secuencial de `UserRecord` y el número subsecuencial se someten a un punto de comprobación.
La llamada a la nueva operación de punto de control de KCL `checkpoint(String sequenceNumber, long subSequenceNumber)` crea explícitamente un punto de control del número secuencial dado junto con el número subsecuencial dado.
En cualquiera de estos casos, después de que el punto de control se almacene en la tabla de Amazon DynamoDB, KCL puede reanudar correctamente la recuperación de los registros, incluso cuando la aplicación se bloquee y se reinicie. Si la secuencia contiene más registros, la recuperación se produce a partir del registro con el siguiente número subsecuencial dentro del registro con el número secuencial que se haya comprobado más recientemente. Si el punto de comprobación más reciente incluyó el último número subsecuencial del registro con el número secuencial anterior, la recuperación se produce a partir del registro con el siguiente número secuencial.
En la siguiente sección se explican los detalles de la comprobación secuencial y subsecuencial para los consumidores que deban evitar la omisión y la duplicación de registros. Si la omisión (o duplicación) de registros al detener y reiniciar el procesamiento de registros del consumidor no resulta importante, puede ejecutar su código existente sin modificaciones.
## Usar extensiones de KCL para la desagrupación de KPL
La desagrupación de KPL puede implicar la creación de puntos de control subsecuenciales. Para facilitar el uso de la creación de puntos de control subsecuenciales, se ha agregado una clase `UserRecord` a KCL:
```
public class UserRecord extends Record {
public long getSubSequenceNumber() {
/* ... */
}
@Override
public int hashCode() {
/* contract-satisfying implementation */
}
@Override
public boolean equals(Object obj) {
/* contract-satisfying implementation */
}
}
```
Esta categoría es la que se usa ahora en lugar de `Record`. Esto no afecta al código existente, porque es una subclase de `Record`. La clase `UserRecord` representa tanto subregistros reales como registros estándares no agregados. Se pueden describir los registros no agregados como registros agregados con solo un subregistro.
Además, se han añadido dos nuevas operaciones a `IRecordProcessorCheckpointer`:
```
public void checkpoint(Record record);
public void checkpoint(String sequenceNumber, long subSequenceNumber);
```
Para empezar a utilizar la creación de puntos de comprobación de números subsecuenciales, puede realizar la siguiente conversión. Cambie el siguiente código de formulario:
```
checkpointer.checkpoint(record.getSequenceNumber());
```
Nuevo código de formulario:
```
checkpointer.checkpoint(record);
```
Le recomendamos que use el formulario `checkpoint(Record record)` para la generación de puntos de comprobación subsecuenciales. Sin embargo, si ya está almacenando `sequenceNumbers` en cadenas para la creación de puntos de comprobación, ahora también deberá almacenar `subSequenceNumber`, tal y como se muestra en el ejemplo siguiente:
```
String sequenceNumber = record.getSequenceNumber();
long subSequenceNumber = ((UserRecord) record).getSubSequenceNumber(); // ... do other processing
checkpointer.checkpoint(sequenceNumber, subSequenceNumber);
```
La difusión de `Record` a `UserRecord` siempre se realiza correctamente, ya que en la implementación siempre se usa `UserRecord`. A no ser que sea necesario realizar cálculos aritméticos en los números secuenciales, no recomendamos este enfoque.
Al procesar los registros de usuario de KPL, KCL escribe el número subsecuencial en Amazon DynamoDB como un campo adicional para cada fila. Las versiones anteriores de KCL utilizaban `AFTER_SEQUENCE_NUMBER` para obtener los registros al reanudar los puntos de control. KCL actual con compatibilidad con KPL utiliza `AT_SEQUENCE_NUMBER` en su lugar. Cuando se recupera el registro del número secuencial del punto de comprobación, se comprueba el número secuencial sometido al punto de comprobación, y los subregistros se descartan según proceda (podrían ser todos ellos, si el último subregistro es el del punto de comprobación). De nuevo, se pueden entender los registros no agregados como registros agregados con un único subregistro, de modo que el mismo algoritmo funciona tanto para los registros agregados como para los no agregados.
## Úselo GetRecords directamente
También puede optar por no utilizar KCL e invocar en su lugar la operación de la API `GetRecords` directamente para recuperar registros de Kinesis Data Streams. Para desempaquetar estos registros recuperados en el registro de usuario original de KPL, llame a una de las siguientes operaciones estáticas en `UserRecord.java`:
```
public static List deaggregate(List records)
public static List deaggregate(List records, BigInteger startingHashKey, BigInteger endingHashKey)
```
La primera operación utiliza el valor predeterminado `0` (cero) para `startingHashKey` y el valor predeterminado `2^128 -1` para `endingHashKey`.
Cada una de estas operaciones desagrupa la lista de registros de Kinesis Data Streams especificada y genera una lista de registros de usuario de KPL. Cualquier registro de usuario de KPL cuya clave hash explícita o clave de partición quede fuera del rango entre `startingHashKey` (incluida) y `endingHashKey` (incluida) se descarta de la lista de registros devuelta.
# Usar KPL con Amazon Data Firehose
Si utiliza Kinesis Producer Library (KPL) para escribir datos en un flujo de datos de Kinesis, puede utilizar la agregación para combinar los registros que escriba en ese flujo de datos de Kinesis. Si después utiliza ese flujo de datos como origen del flujo de entrega de Firehose, este desagrupa los registros antes de entregarlos en el destino. Si configura el flujo de entrega de modo que transforme los datos, Firehose desagrega los registros antes de entregarlos en AWS Lambda. Para obtener más información, consulte [Escritura en Amazon Firehose mediante Kinesis Data Streams](https://docs.aws.amazon.com/firehose/latest/dev/writing-with-kinesis-streams.html).
# Utilice el KPL con el registro de AWS Glue esquemas
Puede integrar sus flujos de datos de Kinesis con el registro de AWS Glue esquemas. AWS Glue Schema Registry le permite descubrir, controlar y evolucionar de forma centralizada esquemas, además de garantizar que un esquema registrado valide de forma continua los datos generados. 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. El registro de AWS Glue esquemas le permite mejorar la end-to-end calidad y el gobierno de los datos en sus aplicaciones de streaming. Para obtener más información, consulte [AWS Glue Schema Registry](https://docs.aws.amazon.com/glue/latest/dg/schema-registry.html). Una de las maneras de configurar esta integración es a través de las bibliotecas KPL y Kinesis Client Library (KCL) en Java.
**importante**
Actualmente, la integración de Kinesis Data Streams AWS Glue y el registro de esquemas solo se admite para las transmisiones de datos de Kinesis que utilizan productores de KPL implementados en Java. No se proporciona soporte multilingüe.
Para obtener instrucciones detalladas sobre cómo configurar la integración de Kinesis Data Streams con Schema Registry mediante la KPL, consulte la sección «Interacción con los datos mediante KPL/KCL las bibliotecas» [en Caso de uso: integración de Amazon Kinesis Data Streams con AWS el](https://docs.aws.amazon.com/glue/latest/dg/schema-registry-integrations.html#schema-registry-integrations-kds) registro de esquemas de Glue.
# Configurar la configuración del proxy de KPL
En el caso de las aplicaciones que no pueden conectarse directamente a Internet, todos los clientes AWS del SDK admiten el uso de proxies HTTP o HTTPS. En un entorno empresarial típico, todo el tráfico de red saliente tiene que pasar por servidores proxy. Si su aplicación utiliza la biblioteca de productores de Kinesis (KPL) para recopilar y enviar datos AWS en un entorno que utiliza servidores proxy, necesitará una configuración de proxy KPL. KPL es una biblioteca de alto nivel creada sobre el SDK de AWS Kinesis. Se divide en un proceso nativo y un contenedor. El proceso nativo realiza todos los trabajos de procesamiento y envío de registros, mientras que el contenedor administra el proceso nativo y se comunica con este. Para obtener más información, consulte [Implementing Efficient and Reliable Producers with the Amazon Kinesis Producer Library](https://aws.amazon.com/blogs/big-data/implementing-efficient-and-reliable-producers-with-the-amazon-kinesis-producer-library/).
El contenedor está escrito en Java y el proceso nativo está escrito en C\$1\$1 con el uso del SDK de Kinesis. La versión 0.14.7 y posterior de KPL ahora admite la configuración del proxy en el contenedor de Java, que puede transferir todas las configuraciones del proxy al proceso nativo. Para obtener más información, consulte [https://github.com/awslabs/amazon-kinesis-producer/releases/tag/v](https://github.com/awslabs/amazon-kinesis-producer/releases/tag/v0.14.7)0.14.7.
Puede utilizar el código siguiente para agregar las configuraciones del proxy a las aplicaciones de KPL.
```
KinesisProducerConfiguration configuration = new KinesisProducerConfiguration();
// Next 4 lines used to configure proxy
configuration.setProxyHost("10.0.0.0"); // required
configuration.setProxyPort(3128); // default port is set to 443
configuration.setProxyUserName("username"); // no default
configuration.setProxyPassword("password"); // no default
KinesisProducer kinesisProducer = new KinesisProducer(configuration);
```
# Política de ciclo de vida de la versión de KPL
En este tema se describe la política de ciclo de vida de las versiones de Amazon Kinesis Producer Library (KPL). AWS publica periódicamente nuevas versiones de las versiones de KPL para incorporar nuevas funciones y mejoras, correcciones de errores, parches de seguridad y actualizaciones de dependencias. Le recomendamos que utilice las versiones KPL para mantenerse al día con las últimas funciones, actualizaciones de seguridad y dependencias subyacentes. up-to-date **No** recomendamos seguir utilizando una versión de KPL no compatible.
El ciclo de vida de las principales versiones de KPL consta de las siguientes tres fases:
+ **Disponibilidad general (GA)**: durante esta fase, la versión principal es totalmente compatible. AWS proporciona versiones periódicas de versiones secundarias y de parches que incluyen compatibilidad con nuevas funciones o actualizaciones de API para Kinesis Data Streams, así como correcciones de errores y de seguridad.
+ **Modo de mantenimiento**: AWS limita las publicaciones de las versiones de parches para abordar únicamente las correcciones de errores críticos y los problemas de seguridad. La versión principal no recibirá actualizaciones de las nuevas funciones ni APIs de Kinesis Data Streams.
+ **E nd-of-support**: La versión principal ya no recibirá actualizaciones ni versiones. Las versiones publicadas anteriormente seguirán estando disponibles a través de los administradores de paquetes públicos y el código permanecerá activado GitHub. El uso de una versión disponible end-of-support queda a discreción del usuario. Recomendamos que actualice a la versión principal más reciente.
| Versión principal | Fase actual | Fecha de publicación | Fecha del modo de mantenimiento | End-of-support fecha |
| --- | --- | --- | --- | --- |
| KPL 0.x | Modo de mantenimiento | 2015-06-02 | 2025-04-17 | 2026-01-30 |
| KPL 1.x | Disponibilidad general | 2024-12-15 | -- | -- |
# Desarrolle productores mediante la API Amazon Kinesis Data Streams con AWS SDK para Java
Puede desarrollar productores mediante la API de Amazon Kinesis Data Streams con AWS el SDK for Java. Si es la primera vez que utiliza Kinesis Data Streams, familiarícese antes con los conceptos y los términos que encontrará en [¿Qué es Amazon Kinesis Data Streams?](introduction.md) y [Úselo AWS CLI para realizar operaciones de Amazon Kinesis Data Streams](getting-started.md).
En estos ejemplos se aborda la [API de Kinesis Data Streams](https://docs.aws.amazon.com/kinesis/latest/APIReference/) y se utiliza el [SDK de AWS para Java](https://aws.amazon.com/sdk-for-java/) para insertar datos en un flujo. Sin embargo, en la mayoría de los casos de uso, es preferible optar por la biblioteca KPL de Kinesis Data Streams. Para obtener más información, consulte [Desarrollo de productores con Amazon Kinesis Producer Library (KPL)](developing-producers-with-kpl.md).
El código de ejemplo de Java de este capítulo demuestra cómo realizar operaciones básicas con la API de Kinesis Data Streams y está dividido lógicamente por tipo de operación. Estos ejemplos no representan códigos listos para producción, ya que no comprueban todas las excepciones posibles ni toman en cuenta todas las consideraciones de seguridad y desempeño posibles. Además, puede llamar a la [API de Kinesis Data Streams](https://docs.aws.amazon.com/kinesis/latest/APIReference/) mediante otros lenguajes de programación. Para obtener más información sobre todos los productos disponibles AWS SDKs, consulte [Comience a desarrollar con Amazon Web Services](https://aws.amazon.com/developers/getting-started/).
Cada tarea tiene requisitos previos. Por ejemplo, no se pueden agregar datos a una secuencia hasta que se haya creado una, para lo que se ha de crear un cliente. Para obtener más información, consulte [Crear y administrar Kinesis Data Streams](working-with-streams.md).
**Topics**
+ [Agregar datos a un flujo](#kinesis-using-sdk-java-add-data-to-stream)
+ [Interactúe con los datos mediante el registro AWS Glue de esquemas](kinesis-integration-glue-schema-registry.md)
## Agregar datos a un flujo
Una vez que se crea una secuencia, puede agregar datos a ella en forma de registros. Un registro es una estructura de datos que contiene los datos que se han de procesar en forma de un blob de datos. Después de almacenar los datos en el registro, Kinesis Data Streams no inspecciona, interpreta ni cambia los datos. Cada registro también tiene asociado un número secuencial y una clave de partición.
Existen dos operaciones diferentes en la API de Kinesis Data Streams que agregan datos a un flujo: [https://docs.aws.amazon.com/kinesis/latest/APIReference/API_PutRecords.html](https://docs.aws.amazon.com/kinesis/latest/APIReference/API_PutRecords.html) y [https://docs.aws.amazon.com/kinesis/latest/APIReference/API_PutRecord.html](https://docs.aws.amazon.com/kinesis/latest/APIReference/API_PutRecord.html). La operación `PutRecords` envía varios registros a su secuencia por solicitud HTTP, y la operación única `PutRecord` envía registros a su secuencia, una por una (se necesita una solicitud HTTP independiente para cada registro). Es preferible utilizar `PutRecords` para la mayoría de las aplicaciones, ya que conseguirá un mayor rendimiento por productor de datos. Para obtener más información sobre cada una de estas operaciones, consulte las subsecciones independientes que aparecen a continuación.
**Topics**
+ [Añada varios registros con PutRecords](#kinesis-using-sdk-java-putrecords)
+ [Añada un único registro con PutRecord](#kinesis-using-sdk-java-putrecord)
Siempre debe tener en mente que, a medida que la aplicación de origen agrega datos al flujo mediante la API de Kinesis Data Streams, muy probablemente haya una o varias aplicaciones de consumidor que procesan datos simultáneamente desde el flujo. Para obtener información sobre cómo los consumidores obtienen datos mediante la API de Kinesis Data Streams, consulte [Obtener datos de un flujo](developing-consumers-with-sdk.md#kinesis-using-sdk-java-get-data).
**importante**
[Cambiar el periodo de retención de datos](kinesis-extended-retention.md)
### Añada varios registros con PutRecords
La operación [https://docs.aws.amazon.com/kinesis/latest/APIReference/API_PutRecords.html](https://docs.aws.amazon.com/kinesis/latest/APIReference/API_PutRecords.html) envía varios registros a Kinesis Data Streams en una única solicitud. Al utilizar `PutRecords`, los productores pueden conseguir un mayor rendimiento cuando envían datos a los flujos de datos de Kinesis. Cada solicitud `PutRecords` puede admitir hasta 500 registros. Cada registro puede ser tan grande como 1 MB, hasta un límite de 5 MB para toda la solicitud, incluidas las claves de partición. Al igual que con la operación única `PutRecord` que se describe a continuación, `PutRecords` utiliza números secuenciales y claves de partición. Sin embargo, el parámetro de `PutRecord` `SequenceNumberForOrdering` no se incluye en una llamada `PutRecords`. La operación `PutRecords` intenta procesar todos los registros en el orden natural de la solicitud.
Cada registro de datos tiene un número secuencial único. Kinesis Data Streams asigna el número secuencial cuando se llama a `client.putRecords` para agregar los registros de datos al flujo. Por lo general, los números secuenciales de una misma clave de partición aumentan con el tiempo; cuanto más grande sea el periodo de tiempo transcurrido entre las solicitudes a `PutRecords`, más aumentan los números secuenciales.
**nota**
Los números secuenciales no se pueden utilizar como índices de conjuntos de datos dentro de la misma secuencia. Para separar lógicamente conjuntos de datos, utilice claves de partición o cree una secuencia independiente para cada conjunto de datos.
Una solicitud `PutRecords` puede incluir registros con diferentes claves de partición. El ámbito de la solicitud es una secuencia; cada solicitud puede incluir cualquier combinación de claves de partición y registros hasta alcanzar los límites de la solicitud. Las solicitudes realizadas con varias claves de partición a secuencias con muchos fragmentos diferentes suelen ser más rápidas que las solicitudes con un número reducido de claves de partición a un número pequeño de fragmentos. El número de claves de partición debe ser mucho mayor que el número de fragmentos a fin de reducir la latencia y maximizar el rendimiento.
#### PutRecords ejemplo
El siguiente código crea 100 registros de datos con claves de partición secuenciales y los inserta en una secuencia llamada `DataStream`.
```
AmazonKinesisClientBuilder clientBuilder = AmazonKinesisClientBuilder.standard();
clientBuilder.setRegion(regionName);
clientBuilder.setCredentials(credentialsProvider);
clientBuilder.setClientConfiguration(config);
AmazonKinesis kinesisClient = clientBuilder.build();
PutRecordsRequest putRecordsRequest = new PutRecordsRequest();
putRecordsRequest.setStreamName(streamName);
List putRecordsRequestEntryList = new ArrayList<>();
for (int i = 0; i < 100; i++) {
PutRecordsRequestEntry putRecordsRequestEntry = new PutRecordsRequestEntry();
putRecordsRequestEntry.setData(ByteBuffer.wrap(String.valueOf(i).getBytes()));
putRecordsRequestEntry.setPartitionKey(String.format("partitionKey-%d", i));
putRecordsRequestEntryList.add(putRecordsRequestEntry);
}
putRecordsRequest.setRecords(putRecordsRequestEntryList);
PutRecordsResult putRecordsResult = kinesisClient.putRecords(putRecordsRequest);
System.out.println("Put Result" + putRecordsResult);
```
La respuesta `PutRecords` incluye una gama de `Records` de respuesta. Cada registro de la matriz de respuestas se correlaciona directamente con un registro en la matriz de solicitudes siguiendo el orden natural, de arriba abajo de la solicitud y la respuesta. La matriz de `Records` de respuesta siempre incluye el mismo número de registros que la matriz de solicitudes.
#### Maneje las fallas al usar PutRecords
De forma predeterminada, el error de registros individuales en una solicitud no para el procesamiento de los registros siguientes en una solicitud `PutRecords`. Esto significa que una matriz de `Records` de respuesta incluye tanto los registros procesados correctamente como los que no. Debe detectar los registros procesados de forma incorrecta e incluirlos en una llamada posterior.
Los registros correctos incluyen los valores `SequenceNumber` y `ShardID`, y los registros incorrectos incluyen los valores `ErrorCode` y `ErrorMessage`. El parámetro `ErrorCode` refleja el tipo de error y puede tomar uno de los siguientes valores: `ProvisionedThroughputExceededException` o `InternalFailure`. `ErrorMessage` proporciona información más detallada sobre la excepción `ProvisionedThroughputExceededException`, e incluye el ID de la cuenta, el nombre de la secuencia y el ID del fragmento del registro al que se ha aplicado limitación. El ejemplo siguiente tiene tres registros en una solicitud `PutRecords`. El segundo registro ha generado un error y se refleja en la respuesta.
**Example PutRecords Solicita la sintaxis**
```
{
"Records": [
{
"Data": "XzxkYXRhPl8w",
"PartitionKey": "partitionKey1"
},
{
"Data": "AbceddeRFfg12asd",
"PartitionKey": "partitionKey1"
},
{
"Data": "KFpcd98*7nd1",
"PartitionKey": "partitionKey3"
}
],
"StreamName": "myStream"
}
```
**Example PutRecords Sintaxis de respuesta**
```
{
"FailedRecordCount”: 1,
"Records": [
{
"SequenceNumber": "21269319989900637946712965403778482371",
"ShardId": "shardId-000000000001"
},
{
“ErrorCode":”ProvisionedThroughputExceededException”,
“ErrorMessage": "Rate exceeded for shard shardId-000000000001 in stream exampleStreamName under account 111111111111."
},
{
"SequenceNumber": "21269319989999637946712965403778482985",
"ShardId": "shardId-000000000002"
}
]
}
```
Los registros que se procesan sin éxito se pueden incluir en las solicitudes `PutRecords` posteriores. En primer lugar, compruebe el parámetro `FailedRecordCount` en `putRecordsResult` para confirmar si se hay registros con error en la solicitud. En caso afirmativo, cada `putRecordsEntry` que tenga un `ErrorCode` que no sea `null` se debe agregar a una solicitud posterior. Para un ejemplo de este tipo de controlador, consulte el siguiente código.
**Example PutRecords manejador de fallas**
```
PutRecordsRequest putRecordsRequest = new PutRecordsRequest();
putRecordsRequest.setStreamName(myStreamName);
List putRecordsRequestEntryList = new ArrayList<>();
for (int j = 0; j < 100; j++) {
PutRecordsRequestEntry putRecordsRequestEntry = new PutRecordsRequestEntry();
putRecordsRequestEntry.setData(ByteBuffer.wrap(String.valueOf(j).getBytes()));
putRecordsRequestEntry.setPartitionKey(String.format("partitionKey-%d", j));
putRecordsRequestEntryList.add(putRecordsRequestEntry);
}
putRecordsRequest.setRecords(putRecordsRequestEntryList);
PutRecordsResult putRecordsResult = amazonKinesisClient.putRecords(putRecordsRequest);
while (putRecordsResult.getFailedRecordCount() > 0) {
final List failedRecordsList = new ArrayList<>();
final List putRecordsResultEntryList = putRecordsResult.getRecords();
for (int i = 0; i < putRecordsResultEntryList.size(); i++) {
final PutRecordsRequestEntry putRecordRequestEntry = putRecordsRequestEntryList.get(i);
final PutRecordsResultEntry putRecordsResultEntry = putRecordsResultEntryList.get(i);
if (putRecordsResultEntry.getErrorCode() != null) {
failedRecordsList.add(putRecordRequestEntry);
}
}
putRecordsRequestEntryList = failedRecordsList;
putRecordsRequest.setRecords(putRecordsRequestEntryList);
putRecordsResult = amazonKinesisClient.putRecords(putRecordsRequest);
}
```
### Añada un único registro con PutRecord
Cada llamada a [https://docs.aws.amazon.com/kinesis/latest/APIReference/API_PutRecord.html](https://docs.aws.amazon.com/kinesis/latest/APIReference/API_PutRecord.html) opera sobre un solo registro. Es preferible recurrir a la operación `PutRecords` que se describe en [Añada varios registros con PutRecords](#kinesis-using-sdk-java-putrecords) a menos que su aplicación necesite específicamente enviar siempre registros individuales en cada solicitud, o que por cualquier otro motivo no se pueda utilizar `PutRecords`.
Cada registro de datos tiene un número secuencial único. Kinesis Data Streams asigna el número secuencial cuando se llama a `client.putRecord` para agregar el registro de datos al flujo. Por lo general, los números secuenciales de una misma clave de partición aumentan con el tiempo; cuanto más grande sea el periodo de tiempo transcurrido entre las solicitudes a `PutRecord`, más aumentan los números secuenciales.
Cuando se producen inserciones (puts) en una sucesión rápida, no hay garantía de que los números secuenciales devueltos aumenten, ya que las operaciones put aparecen esencialmente de manera simultánea a Kinesis Data Streams. Para garantizar unos números secuenciales estrictamente en aumento para la misma clave de partición, utilice el parámetro `SequenceNumberForOrdering` tal y como se muestra en el código de muestra de [PutRecord ejemplo](#kinesis-using-sdk-java-putrecord-example).
Independientemente de que use o no `SequenceNumberForOrdering`, los registros que recibe Kinesis Data Streams mediante una llamada a `GetRecords` están estrictamente ordenados por número secuencial.
**nota**
Los números secuenciales no se pueden utilizar como índices de conjuntos de datos dentro de la misma secuencia. Para separar lógicamente conjuntos de datos, utilice claves de partición o cree una secuencia independiente para cada conjunto de datos.
Una clave de partición se utiliza para agrupar datos dentro de una secuencia. Un registro de datos se asigna a un fragmento dentro de la secuencia en función de su clave de partición. En concreto, Kinesis Data Streams utiliza la clave de partición como entrada para una función hash que asigna la clave de partición (y los datos asociados) a una partición específica.
Como resultado de este mecanismo de hash, todos los registros de datos con la misma clave de partición se asignan al mismo fragmento dentro de la secuencia. Sin embargo, si el número de claves de partición supera el número de fragmentos, algunos fragmentos han de contener necesariamente registros con diferentes claves de partición. Desde el punto de vista del diseño, para garantizar que todos los fragmentos estén bien utilizados, el número de fragmentos (especificado mediante el método `setShardCount` de `CreateStreamRequest`) debe ser significativamente inferior al número de claves de partición únicas, y la cantidad de datos que entran en una única clave de partición debe ser significativamente inferior a la capacidad del fragmento.
#### PutRecord ejemplo
El siguiente código crea diez registros de datos distribuidos en dos claves de partición y los inserta en una secuencia llamada `myStreamName`.
```
for (int j = 0; j < 10; j++)
{
PutRecordRequest putRecordRequest = new PutRecordRequest();
putRecordRequest.setStreamName( myStreamName );
putRecordRequest.setData(ByteBuffer.wrap( String.format( "testData-%d", j ).getBytes() ));
putRecordRequest.setPartitionKey( String.format( "partitionKey-%d", j/5 ));
putRecordRequest.setSequenceNumberForOrdering( sequenceNumberOfPreviousRecord );
PutRecordResult putRecordResult = client.putRecord( putRecordRequest );
sequenceNumberOfPreviousRecord = putRecordResult.getSequenceNumber();
}
```
El código de muestra anterior utiliza `setSequenceNumberForOrdering` para garantizar un orden estrictamente creciente en cada clave de partición. Para utilizar este parámetro de forma eficaz, establezca como `SequenceNumberForOrdering` del registro actual (registro *n*) el número secuencial del registro anterior (registro *n-1*). Para obtener el número secuencial de un registro que se ha añadido a la secuencia, llame a `getSequenceNumber` en el resultado de `putRecord`.
El parámetro `SequenceNumberForOrdering` garantiza números de secuencia estrictamente crecientes para la misma clave de partición. `SequenceNumberForOrdering` no permite ordenar los registros en varias claves de partición.
# Interactúe con los datos mediante el registro AWS Glue de esquemas
Puede integrar sus flujos de datos de Kinesis con el registro de AWS Glue esquemas. AWS Glue Schema Registry le permite descubrir, controlar y evolucionar de forma centralizada esquemas, además de garantizar que un esquema registrado valide de forma continua los datos generados. 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. El registro AWS Glue de esquemas le permite mejorar la end-to-end calidad y el gobierno de los datos en sus aplicaciones de streaming. Para obtener más información, consulte [AWS Glue Schema Registry](https://docs.aws.amazon.com/glue/latest/dg/schema-registry.html). Una de las formas de configurar esta integración es mediante Kinesis Data Streams `PutRecords` y `PutRecord` Kinesis Data APIs Streams disponibles en AWS el SDK de Java.
Para obtener instrucciones detalladas sobre cómo configurar la integración de Kinesis Data Streams con el registro de esquemas mediante Kinesis Data Streams PutRecord y Kinesis, consulte PutRecords la [sección «Interacción con los datos APIs mediante APIs Kinesis Data Streams» en Caso de uso: integración de Amazon Kinesis Data Streams con el registro de esquemas de Glue](https://docs.aws.amazon.com/glue/latest/dg/schema-registry-integrations.html#schema-registry-integrations-kds). AWS
# Escrivir en Amazon Kinesis Data Streams mediante el agente de Kinesis
El agente de Kinesis es una aplicación de software Java independiente que ofrece una forma sencilla de recopilar y enviar datos a Kinesis Data Streams. El agente monitoriza constantemente un conjunto de archivos y envía nuevos datos a su secuencia. El agente se encarga de rotar los archivos, crear puntos de restauración y realizar reintentos cuando se producen errores. El agente entrega todos los datos de manera confiable, puntual y sencilla. También emite CloudWatch métricas de Amazon para ayudarte a supervisar y solucionar mejor los problemas del proceso de streaming.
De forma predeterminada, los registros de cada archivo se analizan en función del carácter de nueva línea (`'\n'`). Sin embargo, el agente también se puede configurar para analizar registros multilínea (consulte [Especificar las opciones de configuración del agente](#agent-config-settings)).
Puede instalar el agente en entornos de servidor basados en Linux, como servidores web, de registro o de base de datos. Después de instalar el agente, configúrelo especificando los archivos que desee monitorizar y la secuencia de los datos. Una vez configurado, el agente recopila datos de los archivos de forma duradera y los envía de forma confiable a la secuencia.
**Topics**
+ [Cumplir con los requisitos previos para el agente de Kinesis](#prereqs)
+ [Descargar e instalar el agente](#download-install)
+ [Configurar e iniciar el agente](#config-start)
+ [Especificar las opciones de configuración del agente](#agent-config-settings)
+ [Monitorear varios directorios de archivos y escribir en varias secuencias](#sim-writes)
+ [Usar el agente para preprocesar el agente](#pre-processing)
+ [Usar los comandos de la CLI](#cli-commands)
+ [Preguntas frecuentes](#agent-faq)
## Cumplir con los requisitos previos para el agente de Kinesis
+ Su sistema operativo debe ser la versión 2015.09 o posterior de Amazon Linux AMI o la versión 7 o posterior de Red Hat Enterprise Linux.
+ Si utiliza Amazon EC2, lance la instancia de EC2 para ejecutar el agente.
+ Gestiona tus AWS credenciales mediante uno de los siguientes métodos:
+ Especifique un rol de IAM al lanzar la instancia EC2.
+ Especifique AWS las credenciales al configurar el agente (consulte [awsAccessKeyID](#awsAccessKeyId) y [awsSecretAccessclave](#awsSecretAccessKey)).
+ `/etc/sysconfig/aws-kinesis-agent`Edítelo para especificar su región y sus claves de AWS acceso.
+ [[Si su instancia EC2 está en una AWS cuenta diferente, cree una función de IAM para proporcionar acceso al servicio Kinesis Data Streams y especifique esa función al configurar el agente (consulte AssumeroLearn and Id). assumeRoleExternal](#assumeRoleExternalId)](#assumeRoleARN) Utilice uno de los métodos anteriores para especificar AWS las credenciales de un usuario de la otra cuenta que tenga permiso para asumir este rol.
+ El rol o AWS las credenciales de IAM que especifique deben tener permiso para realizar la operación de Kinesis Data [PutRecords](https://docs.aws.amazon.com/kinesis/latest/APIReference/API_PutRecords.html)Streams para que el agente envíe datos a su transmisión. Si habilita la CloudWatch supervisión del agente, también necesitará permiso para realizar la CloudWatch [PutMetricData](https://docs.aws.amazon.com/AmazonCloudWatch/latest/APIReference/API_PutMetricData.html)operación. Para obtener más información, consulte [Control del acceso a los recursos de Amazon Kinesis Data Streams mediante IAM](controlling-access.md)[Supervise el estado de los agentes de Kinesis Data Streams con Amazon CloudWatch](agent-health.md), y [Control de CloudWatch acceso](https://docs.aws.amazon.com/AmazonCloudWatch/latest/DeveloperGuide/UsingIAM.html).
## Descargar e instalar el agente
Primero, conéctese a la instancia. Para obtener más información, consulte [Conexión a una instancia](https://docs.aws.amazon.com/AWSEC2/latest/UserGuide/ec2-connect-to-instance-linux.html) en la *Guía del usuario de Amazon EC2*. Si tiene problemas para conectarse, consulte [Solución de problemas con la conexión a la instancia](https://docs.aws.amazon.com/AWSEC2/latest/UserGuide/TroubleshootingInstancesConnecting.html) en la *Guía del usuario de Amazon EC2*.
**Para configurar el agente a través de la AMI de Amazon Linux**
Use el siguiente comando para descargar e instalar el agente:
```
sudo yum install –y aws-kinesis-agent
```
**Para configurar el agente en Red Hat Enterprise Linux**
Use el siguiente comando para descargar e instalar el agente:
```
sudo yum install –y https://s3.amazonaws.com/streaming-data-agent/aws-kinesis-agent-latest.amzn2.noarch.rpm
```
**Para configurar el agente mediante GitHub**
1. Descargue el agente desde [awlabs/ amazon-kinesis-agent](https://github.com/awslabs/amazon-kinesis-agent).
1. Instale el agente. Para ello, diríjase al directorio de descargas y ejecute el siguiente comando:
```
sudo ./setup --install
```
**Configuración del agente en un contenedor de Docker**
El agente de Kinesis también puede ejecutarse en un contenedor además de a través de la base de contenedores [amazonlinux](https://docs.aws.amazon.com/AmazonECR/latest/userguide/amazon_linux_container_image.html). Utilice el siguiente Dockerfile y ejecute `docker build`.
```
FROM amazonlinux
RUN yum install -y aws-kinesis-agent which findutils
COPY agent.json /etc/aws-kinesis/agent.json
CMD ["start-aws-kinesis-agent"]
```
## Configurar e iniciar el agente
**Configuración e inicio del agente**
1. Abra y edite el archivo de configuración (como superusuario si utiliza permisos de acceso de archivo predeterminado): `/etc/aws-kinesis/agent.json`
En este archivo de configuración, especifique los archivos (`"filePattern"`) desde los que el agente deberá recopilar datos y el nombre de la secuencia (`"kinesisStream"`) a la que deberá enviarlos. Tenga en cuenta que el nombre de archivo es un patrón y que el agente reconoce rotaciones de archivos. No puede rotar más de un archivo ni crear más de uno nuevo por segundo. El agente utiliza la marca temporal de creación de archivos para determinar a qué archivos realizarles seguimiento y poner en fila en la secuencia. Crear nuevos archivos o rotarlos más de una vez por segundo evita que el agente pueda diferenciarlos correctamente.
```
{
"flows": [
{
"filePattern": "/tmp/app.log*",
"kinesisStream": "yourkinesisstream"
}
]
}
```
1. Comience el agente de forma manual:
```
sudo service aws-kinesis-agent start
```
1. Configure el agente para iniciarse al arrancar el sistema (opcional):
```
sudo chkconfig aws-kinesis-agent on
```
El agente ya está se ejecutando como un servicio de sistema en segundo plano. Monitoriza constantemente los archivos especificados y envía datos a la secuencia especificada. La auditoría de actividad se registra en `/var/log/aws-kinesis-agent/aws-kinesis-agent.log`.
## Especificar las opciones de configuración del agente
El agente admite las dos opciones de configuración obligatorias, `filePattern` y `kinesisStream`, además de configuraciones opcionales para activar características adicionales. Las opciones de configuración obligatorias y opcionales se especifican en `/etc/aws-kinesis/agent.json`.
Cada vez que cambie el archivo de configuración, debe detener y comenzar el agente con los siguientes comandos:
```
sudo service aws-kinesis-agent stop
sudo service aws-kinesis-agent start
```
También puede hacerlo con el siguiente comando:
```
sudo service aws-kinesis-agent restart
```
Las opciones de configuración generales son las siguientes.
| Opción de configuración | Description (Descripción) |
| --- | --- |
| assumeRoleARN | El ARN del rol que asumirá el usuario. *Para obtener más información, consulte [Delegar el acceso a todas AWS las cuentas mediante funciones de IAM](https://docs.aws.amazon.com/IAM/latest/UserGuide/tutorial_cross-account-with-roles.html) en la Guía del usuario de IAM.* |
| assumeRoleExternalId | Un identificador opcional que determina quién puede asumir el rol. Para obtener más información, consulte [Cómo utilizar un ID externo](https://docs.aws.amazon.com/IAM/latest/UserGuide/id_roles_create_for-user_externalid.html) en la *Guía del usuario de IAM*. |
| awsAccessKeyId | AWS ID de clave de acceso que anula las credenciales predeterminadas. Este ajuste tiene prioridad sobre los demás proveedores de credenciales. |
| awsSecretAccessKey | AWS clave secreta que anula las credenciales predeterminadas. Este ajuste tiene prioridad sobre los demás proveedores de credenciales. |
| cloudwatch.emitMetrics | Permite que el agente emita métricas CloudWatch si se ha establecido (true). Predeterminado: true |
| cloudwatch.endpoint | El punto final regional de CloudWatch. Valor predeterminado: `monitoring.us-east-1.amazonaws.com` |
| kinesis.endpoint | El punto de conexión regional para Kinesis Data Streams. Valor predeterminado: `kinesis.us-east-1.amazonaws.com` |
Las opciones de configuración de flujo son las siguientes.
| Opción de configuración | Description (Descripción) |
| --- | --- |
| dataProcessingOptions | La lista de opciones de procesamiento aplicadas a cada registro analizado antes de enviarlo a la secuencia. Las opciones de procesamiento se realizan en el orden especificado. Para obtener más información, consulte [Usar el agente para preprocesar el agente](#pre-processing). |
| kinesisStream | [Obligatorio] El nombre de la secuencia. |
| filePattern | [Obligatorio] El patrón del directorio y del archivo que debe coincidir para que el agente lo capte. Se debe conceder permisos de lectura a `aws-kinesis-agent-user` para todos los archivos que coincidan con este patrón. Se debe conceder permisos de lectura y ejecución a `aws-kinesis-agent-user` para el directorio que contiene los archivos. |
| initialPosition | La posición inicial desde la que el archivo comenzó a ser analizado. Los valores válidos son `START_OF_FILE` y `END_OF_FILE`. Valor predeterminado: `END_OF_FILE` |
| maxBufferAgeMillis | El tiempo máximo, en milisegundos, durante el cual el agente almacena los datos en búfer antes de enviarlos a la secuencia. Intervalo de valores: 1000 a 900 000 (1 segundo a 15 minutos) Valor predeterminado: 60 000 (1 minuto) |
| maxBufferSizeBytes | La cantidad de datos en bytes que el agente almacena en búfer antes de enviarlos a la secuencia. Intervalo de valores: 1 a 4 194 304 (4 MB) Valor predeterminado: 4 194 304 (4 MB) |
| maxBufferSizeRecords | La cantidad máxima de registros en datos que el agente almacena en búfer antes de enviarlos a la secuencia. Intervalo de valores: 1 a 500 Predeterminado: 500 |
| minTimeBetweenFilePollsMillis | El intervalo de tiempo, en milisegundos, en el que el agente sondea y analiza los archivos monitorizados para identificar datos nuevos. Intervalo de valores: 1 o más Predeterminado: 100 |
| multiLineStartPattern | El patrón para identificar el comienzo de un registro. Un registro consta de una línea que coincide con el patrón y de líneas siguientes que no coinciden con el patrón. Los valores válidos son expresiones regulares. De forma predeterminada, cada línea en los archivos de registro se analiza como un registro. |
| partitionKeyOption | El método para generar la clave de partición. Los valores válidos son `RANDOM` (entero generado aleatoriamente) y `DETERMINISTIC` (un valor hash calculado a partir de los datos). Valor predeterminado: `RANDOM` |
| skipHeaderLines | La cantidad de líneas de los archivos monitorizados, a partir de la primera, que el agente debe omitir en el momento de analizarlos. Intervalo de valores: 0 o más Cantidad predeterminada: 0 (cero) |
| truncatedRecordTerminator | La cadena que utiliza el agente para truncar un registro analizado cuando su tamaño supera el límite de tamaño de registros de Kinesis Data Streams. (1000 KB) Valor predeterminado: `'\n'` (línea nueva) |
## Monitorear varios directorios de archivos y escribir en varias secuencias
Puede configurar el agente para que monitorice varios directorios de archivos y envíe datos a varias secuencias especificando varias opciones de configuración de secuencia. En el siguiente ejemplo de configuración, el agente supervisa dos directorios de archivos y envía datos a un flujo de Kinesis y a un flujo de entrega de Firehose, respectivamente. Tenga en cuenta que puede especificar diferentes puntos de conexión para Kinesis Data Streams y Firehose para que su flujo de Kinesis y su flujo de entrega de Firehose no tengan que estar en la misma región.
```
{
"cloudwatch.emitMetrics": true,
"kinesis.endpoint": "https://your/kinesis/endpoint",
"firehose.endpoint": "https://your/firehose/endpoint",
"flows": [
{
"filePattern": "/tmp/app1.log*",
"kinesisStream": "yourkinesisstream"
},
{
"filePattern": "/tmp/app2.log*",
"deliveryStream": "yourfirehosedeliverystream"
}
]
}
```
Para más información sobre el uso del agente con Firehose, consulte [Escritura en Amazon Data Firehose con el agente de Kinesis](https://docs.aws.amazon.com/firehose/latest/dev/writing-with-agents.html).
## Usar el agente para preprocesar el agente
El agente puede preprocesar los registros previamente analizados de los archivos monitorizados antes de enviarlos a la secuencia. Para habilitar esta característica, añada la opción de configuración `dataProcessingOptions` al flujo de archivos. Se pueden agregar una o más opciones de procesamiento, que se ejecutarán en el orden especificado.
El agente es compatible con las siguientes opciones de procesamiento. Al ser de código abierto, puede desarrollar y ampliar las opciones de procesamiento del agente. Puede descargar el agente desde el [agente de Kinesis](https://github.com/awslabs/amazon-kinesis-agent).Opciones de procesamiento
`SINGLELINE`
Elimina caracteres de nueva línea y espacios al principio y al final de las líneas para convertir un registro multilínea en un registro de una sola línea.
```
{
"optionName": "SINGLELINE"
}
```
`CSVTOJSON`
Convierte un registro de formato con separaciones mediante delimitadores en un registro de formato JSON.
```
{
"optionName": "CSVTOJSON",
"customFieldNames": [ "field1", "field2", ... ],
"delimiter": "yourdelimiter"
}
```
`customFieldNames`
[Obligatorio] Los nombres de campos utilizados como claves en cada par de valores de clave JSON. Por ejemplo, si especifica `["f1", "f2"]`, el registro "v1, v2" se convertirá en `{"f1":"v1","f2":"v2"}`.
`delimiter`
La cadena utilizada como delimitador en el registro. El valor predeterminado es una coma (,).
`LOGTOJSON`
Convierte un registro con un formato de registro en un registro con formato JSON. Los formatos de registro admitidos son **Apache Common Log**, **Apache Combined Log**, **Apache Error Log** y **RFC3164 Syslog**.
```
{
"optionName": "LOGTOJSON",
"logFormat": "logformat",
"matchPattern": "yourregexpattern",
"customFieldNames": [ "field1", "field2", … ]
}
```
`logFormat`
[Obligatorio] El formato de entrada del registro. Los valores posibles son los siguientes:
+ `COMMONAPACHELOG`: formato común de registro de Apache. Cada entrada de registro sigue el siguiente patrón de forma predeterminada: "`%{host} %{ident} %{authuser} [%{datetime}] \"%{request}\" %{response} %{bytes}`".
+ `COMBINEDAPACHELOG`: formato combinado de registro de Apache. Cada entrada de registro sigue el siguiente patrón de forma predeterminada: "`%{host} %{ident} %{authuser} [%{datetime}] \"%{request}\" %{response} %{bytes} %{referrer} %{agent}`".
+ `APACHEERRORLOG`: formato de registro de errores de Apache. Cada entrada de registro sigue el siguiente patrón de forma predeterminada: "`[%{timestamp}] [%{module}:%{severity}] [pid %{processid}:tid %{threadid}] [client: %{client}] %{message}`".
+ `SYSLOG`— El formato RFC3164 Syslog. Cada entrada de registro sigue el siguiente patrón de forma predeterminada: "`%{timestamp} %{hostname} %{program}[%{processid}]: %{message}`".
`matchPattern`
El patrón de expresiones regulares utilizado para extraer valores de entradas de registro. Este ajuste se utiliza si la entrada de registro no tiene uno de los formatos de registro predefinidos. Si se utiliza este ajuste, también tendrá que especificar `customFieldNames`.
`customFieldNames`
Los nombres de campos utilizados como claves en cada par de valores de clave JSON. Utilice esta opción para definir nombres de campos para valores extraídos de `matchPattern`, o sobrescriba los nombres de campos de los formatos de logs predefinidos.
**Example : Configuración LOGTOJSON**
Este es un ejemplo de configuración `LOGTOJSON` de una entrada de registro en Formato común de registro de Apache convertida a formato JSON:
```
{
"optionName": "LOGTOJSON",
"logFormat": "COMMONAPACHELOG"
}
```
Antes de la conversión:
```
64.242.88.10 - - [07/Mar/2004:16:10:02 -0800] "GET /mailman/listinfo/hsdivision HTTP/1.1" 200 6291
```
Después de la conversión:
```
{"host":"64.242.88.10","ident":null,"authuser":null,"datetime":"07/Mar/2004:16:10:02 -0800","request":"GET /mailman/listinfo/hsdivision HTTP/1.1","response":"200","bytes":"6291"}
```
**Example : Configuración LOGTOJSON con campos personalizados**
Este es otro ejemplo de configuración `LOGTOJSON`:
```
{
"optionName": "LOGTOJSON",
"logFormat": "COMMONAPACHELOG",
"customFieldNames": ["f1", "f2", "f3", "f4", "f5", "f6", "f7"]
}
```
Con esta configuración, la misma entrada de registro con Formato común de registro de Apache del ejemplo anterior se convierte a formato JSON de la siguiente manera:
```
{"f1":"64.242.88.10","f2":null,"f3":null,"f4":"07/Mar/2004:16:10:02 -0800","f5":"GET /mailman/listinfo/hsdivision HTTP/1.1","f6":"200","f7":"6291"}
```
**Example : Convertir una entrada de registro con Formato común de registro de Apache**
La siguiente configuración de flujo convierte la entrada de registro con Formato común de registro de Apache en un registro de una línea en formato JSON:
```
{
"flows": [
{
"filePattern": "/tmp/app.log*",
"kinesisStream": "my-stream",
"dataProcessingOptions": [
{
"optionName": "LOGTOJSON",
"logFormat": "COMMONAPACHELOG"
}
]
}
]
}
```
**Example : Convertir registros multilínea**
La siguiente configuración de flujo analiza aquellos registros multilínea cuya primera línea comience por "`[SEQUENCE=`". Primero, el registro se convierte en un registro de una línea. Después, se extraen los valores del registro basándose en tabulaciones delimitadoras. Finalmente, los valores extraídos se asignan a valores `customFieldNames` específicos para formar un registro de una línea en formato JSON.
```
{
"flows": [
{
"filePattern": "/tmp/app.log*",
"kinesisStream": "my-stream",
"multiLineStartPattern": "\\[SEQUENCE=",
"dataProcessingOptions": [
{
"optionName": "SINGLELINE"
},
{
"optionName": "CSVTOJSON",
"customFieldNames": [ "field1", "field2", "field3" ],
"delimiter": "\\t"
}
]
}
]
}
```
**Example : Configuración LOGTOJSON con patrón de coincidencia**
Este es un ejemplo de una configuración de entrada de registro con Formato común de registro de Apache `LOGTOJSON` convertida a formato JSON con el último campo (bytes) omitido:
```
{
"optionName": "LOGTOJSON",
"logFormat": "COMMONAPACHELOG",
"matchPattern": "^([\\d.]+) (\\S+) (\\S+) \\[([\\w:/]+\\s[+\\-]\\d{4})\\] \"(.+?)\" (\\d{3})",
"customFieldNames": ["host", "ident", "authuser", "datetime", "request", "response"]
}
```
Antes de la conversión:
```
123.45.67.89 - - [27/Oct/2000:09:27:09 -0400] "GET /java/javaResources.html HTTP/1.0" 200
```
Después de la conversión:
```
{"host":"123.45.67.89","ident":null,"authuser":null,"datetime":"27/Oct/2000:09:27:09 -0400","request":"GET /java/javaResources.html HTTP/1.0","response":"200"}
```
## Usar los comandos de la CLI
Iniciar automáticamente al agente al arrancar el sistema:
```
sudo chkconfig aws-kinesis-agent on
```
Compruebe el estado del agente:
```
sudo service aws-kinesis-agent status
```
Detener el agente:
```
sudo service aws-kinesis-agent stop
```
Leer el archivo de registro del agente desde esta ubicación:
```
/var/log/aws-kinesis-agent/aws-kinesis-agent.log
```
Desinstalar el agente:
```
sudo yum remove aws-kinesis-agent
```
## Preguntas frecuentes
### ¿Hay un agente de Kinesis para Windows?
El [agente de Kinesis para Windows](https://docs.aws.amazon.com/kinesis-agent-windows/latest/userguide/what-is-kinesis-agent-windows.html) es un software diferente del agente de Kinesis para plataformas Linux.
### ¿Por qué aumenta la velocidad and/or `RecordSendErrors` de Kinesis Agent?
Normalmente esto se debe a la limitación de Kinesis. Compruebe la métrica `WriteProvisionedThroughputExceeded` para Kinesis Data Streams o la métrica `ThrottledRecords` para flujos de entrega de Firehose. Cualquier aumento a partir de 0 en estas métricas indica que es necesario aumentar los límites de flujo. Para obtener más información, consulte [Límites de flujos de datos de Kinesis](https://docs.aws.amazon.com/streams/latest/dev/service-sizes-and-limits.html) y [flujos de entrega de Amazon Firehose](https://docs.aws.amazon.com/firehose/latest/dev/limits.html).
Una vez que descarte la limitación de velocidad, compruebe si el agente de Kinesis está configurado para enviar a la cola una gran cantidad de archivos pequeños. Se produce un retraso cuando el agente de Kinesis sigue un archivo nuevo, por lo que el agente de Kinesis debería seguir una pequeña cantidad de archivos de mayor tamaño. Intente consolidar los archivos de registro en archivos más grandes.
### ¿Por qué se producen excepciones `java.lang.OutOfMemoryError`?
El agente de Kinesis no tiene memoria suficiente para gestionar la carga de trabajo actual. Intente aumentar `JAVA_START_HEAP` y `JAVA_MAX_HEAP` en `/usr/bin/start-aws-kinesis-agent` y reinicie el agente.
### ¿Por qué se producen excepciones `IllegalStateException : connection pool shut down`?
El agente de Kinesis no tiene suficientes conexiones para gestionar la carga de trabajo actual. Intente aumentar `maxConnections` y `maxSendingThreads` en los ajustes generales de configuración del agente en `/etc/aws-kinesis/agent.json`. El valor predeterminado para estos campos es 12 veces los procesadores de tiempo de ejecución disponibles. Consulte [AgentConfiguration.java](https://github.com/awslabs/amazon-kinesis-agent/blob/master/src/com/amazon/kinesis/streaming/agent/config/AgentConfiguration.java) para obtener más información sobre las configuraciones avanzadas de los agentes.
### ¿Cómo puedo depurar otro problema con el agente de Kinesis?
Los registros de nivel `DEBUG` pueden habilitarse en `/etc/aws-kinesis/log4j.xml`.
### ¿Cómo debo configurar el agente de Kinesis?
Cuanto menor sea el valor de `maxBufferSizeBytes`, más frecuentemente enviará datos el agente de Kinesis. Esto puede ser bueno ya que disminuye el tiempo de entrega de los registros, pero también aumenta las solicitudes por segundo a Kinesis.
### ¿Por qué el agente de Kinesis envía registros duplicados?
Esto ocurre debido a una mala configuración en el seguimiento de archivos. Asegúrese de que cada `fileFlow’s filePattern` solo coincida con un archivo. Esto también puede ocurrir si el modo `logrotate` que se está utilizando está en modo `copytruncate`. Intente cambiar al modo predeterminado o al de creación para evitar la duplicación. Para obtener más información sobre la gestión de registros duplicados, consulte [Handling Duplicate Records](https://docs.aws.amazon.com/streams/latest/dev/kinesis-record-processor-duplicates.html).
# Escriba en Kinesis Data Streams mediante otros servicios AWS
Los siguientes AWS servicios se pueden integrar directamente con Amazon Kinesis Data Streams para escribir datos en las transmisiones de datos de Kinesis. Revise la información de cada servicio que le interese y consulte las referencias proporcionadas.
**Topics**
+ [Escriba en Kinesis Data Streams mediante AWS Amplify](using-other-services-amplify.md)
+ [Escribir en Kinesis Data Streams mediante Amazon Aurora](using-other-services-aurora.md)
+ [Escriba en Kinesis Data Streams mediante Amazon CloudFront](using-other-services-CloudFront.md)
+ [Escribir en Kinesis Data Streams mediante Amazon Logs CloudWatch](using-other-services-cw-logs.md)
+ [Escribir en Kinesis Data Streams mediante Amazon Connect](using-other-services-connect.md)
+ [Escriba en Kinesis Data Streams mediante AWS Database Migration Service](using-other-services-migration.md)
+ [Escribir en Kinesis Data Streams mediante Amazon DynamoDB](using-other-services-ddb.md)
+ [Escriba en Kinesis Data Streams mediante Amazon EventBridge](using-other-services-eventbridges.md)
+ [Escriba en Kinesis Data Streams mediante AWS IoT Core](using-other-services-iot-core.md)
+ [Escribir en Kinesis Data Streams mediante Amazon Relational Database Service](using-other-services-rds.md)
+ [Escribir en Kinesis Data Streams mediante Amazon Pinpoint](using-other-services-pinpoint.md)
+ [Escribir en Kinesis Data Streams mediante Amazon Quantum Ledger Database (Amazon QLDB)](using-other-services-quantum-ledger.md)
# Escriba en Kinesis Data Streams mediante AWS Amplify
Puede utilizar Amazon Kinesis Data Streams para transmitir fácilmente datos desde sus aplicaciones móviles creadas con AWS Amplify para procesamiento en tiempo real. A continuación, puede crear paneles de control en tiempo real, capturar excepciones y generar alertas, impulsar recomendaciones y tomar otras decisiones empresariales u operativas en tiempo real. También puede enviar datos fácilmente a otros servicios como Simple Amazon Simple Storage Service, Amazon DynamoDB y Amazon Redshift.
Para obtener más información, consulte [Uso de Amazon Kinesis](https://docs.amplify.aws/react/build-a-backend/more-features/analytics/streaming-data/) en el *Centro para desarrolladores de AWS Amplify*.
# Escribir en Kinesis Data Streams mediante Amazon Aurora
Puede utilizar Amazon Kinesis Data Streams para supervisar las actividades de los clústeres de base de datos de Amazon Aurora. Con Database Activity Streams, el clúster de base de datos de Aurora envía actividades a Amazon Kinesis Data Streams en tiempo real. A continuación, puede crear aplicaciones para la administración de la conformidad que consuman estas actividades, las auditen y generen alertas. También puede utilizar Amazon Firehose para almacenar los datos.
Para obtener más información, consulte [Database Activity Streams](https://docs.aws.amazon.com/AmazonRDS/latest/AuroraUserGuide/DBActivityStreams.html) en la *Guía para desarrolladores de Amazon Aurora*.
# Escriba en Kinesis Data Streams mediante Amazon CloudFront
Puede utilizar Amazon Kinesis Data Streams CloudFront con registros en tiempo real y obtener información sobre las solicitudes realizadas a una distribución en tiempo real. A continuación, puede crear su propio [consumidor de transmisión de datos de Kinesis](https://docs.aws.amazon.com/streams/latest/dev/building-consumers.html) o utilizar Amazon Data Firehose para enviar los datos de registro a Amazon S3, Amazon Redshift, Amazon OpenSearch Service o un servicio de procesamiento de registros de terceros.
Para obtener más información, consulta [Registros en tiempo real](https://docs.aws.amazon.com/AmazonCloudFront/latest/DeveloperGuide/real-time-logs.html) en la *Guía para CloudFront desarrolladores de Amazon*.
# Escribir en Kinesis Data Streams mediante Amazon Logs CloudWatch
Puede utilizar CloudWatch las suscripciones para acceder a un feed en tiempo real de los eventos de registro de Amazon CloudWatch Logs y hacer que se envíen a una transmisión de datos de Kinesis para su procesamiento, análisis y carga en otros sistemas.
Para obtener más información, consulte [Procesamiento de datos de registro en tiempo real con suscripciones](https://docs.aws.amazon.com/AmazonCloudWatch/latest/logs/Subscriptions.html) en la *Guía del usuario de Amazon CloudWatch Logs*.
# Escribir en Kinesis Data Streams mediante Amazon Connect
Puede utilizar Kinesis Data Streams para exportar registros de contactos y eventos de agentes en tiempo real desde su instancia de Amazon Connect. También puede habilitar el flujo de datos desde Perfiles de clientes de Amazon Connect para recibir automáticamente actualizaciones en un flujo de datos de Kinesis sobre la creación de nuevos perfiles o cambios en los existentes.
A continuación, puede crear aplicaciones de consumo para procesar y analizar los datos en tiempo real. Por ejemplo, al utilizar los registros de contactos y los datos del perfil de los clientes, puede conservar los datos de sus sistemas de origen, como CRMs las herramientas de automatización de marketing, up-to-date con la información más reciente. Con los datos de eventos de los agentes, puede crear cuadros de mando que muestren la información y los eventos de los agentes, y activar notificaciones personalizadas de actividades específicas de los agentes.
Para obtener más información, consulte [flujos de datos para su instancia](https://docs.aws.amazon.com/connect/latest/adminguide/data-streaming.html), [configuración de la exportación en tiempo real](https://docs.aws.amazon.com/connect/latest/adminguide/set-up-real-time-export.html) y [flujos de eventos de agentes](https://docs.aws.amazon.com/connect/latest/adminguide/agent-event-streams.html) en la *Guía del administrador de Amazon Connect*.
# Escriba en Kinesis Data Streams mediante AWS Database Migration Service
Puede utilizarlos AWS Database Migration Service para migrar datos a una transmisión de datos de Kinesis. A continuación, puede crear aplicaciones consumidoras que procesen los registros de datos en tiempo real. También puede enviar fácilmente datos a otros servicios como Amazon Simple Storage Service, Amazon DynamoDB y Amazon Redshift.
Para obtener más información, consulte [Uso de Kinesis Data Streams](https://docs.aws.amazon.com/dms/latest/userguide/CHAP_Target.Kinesis.html) en la *Guía del usuario de AWS Database Migration Service *.
# Escribir en Kinesis Data Streams mediante Amazon DynamoDB
Puede utilizar Amazon Kinesis Data Streams para capturar cambios en Amazon DynamoDB. Kinesis Data Streams captura modificaciones a nivel de elemento en cualquier tabla de DynamoDB y las replica en una secuencia de datos de Kinesis. Sus aplicaciones consumidoras pueden obtener acceso a esta transmisión para ver los cambios a nivel de elemento en tiempo real y entregar dichos cambios a los usuarios o tomar medidas en función del contenido.
Para obtener más información, consulte [cómo funciona Kinesis Data Streams con DynamoDB](https://docs.aws.amazon.com/amazondynamodb/latest/developerguide/kds.html) en la *Guía para desarrolladores de Amazon DynamoDB*.
# Escriba en Kinesis Data Streams mediante Amazon EventBridge
Con Kinesis Data Streams, puede AWS enviar eventos EventBridge de [llamadas](https://docs.aws.amazon.com/eventbridge/latest/userguide/eb-events.html) a la API a una transmisión, crear aplicaciones de consumo y procesar grandes cantidades de datos. También puede utilizar Kinesis Data Streams como destino EventBridge en Pipes y entregar los registros en una transmisión desde una de las fuentes disponibles tras el filtrado y el enriquecimiento opcionales.
Para obtener más información, consulte [Enviar eventos a una transmisión de Amazon Kinesis](https://docs.aws.amazon.com/eventbridge/latest/userguide/eb-relay-events-kinesis-stream.html) y [EventBridge Pipes](https://docs.aws.amazon.com/eventbridge/latest/userguide/eb-pipes.html) en la Guía * EventBridge del usuario de Amazon*.
# Escriba en Kinesis Data Streams mediante AWS IoT Core
Puede escribir datos en tiempo real a partir de mensajes MQTT en AWS IoT Core mediante acciones de reglas de AWS IoT. A continuación, puede crear aplicaciones que procesen los datos, analicen su contenido y generen alertas, y entregarlos en aplicaciones de análisis u otros servicios de AWS .
Para obtener más información, consulte [Kinesis Data Streams](https://docs.aws.amazon.com/iot/latest/developerguide/kinesis-rule-action.html) en la *Guía para desarrolladores de AWS IoT Core*.
# Escribir en Kinesis Data Streams mediante Amazon Relational Database Service
Puede utilizar Amazon Kinesis Data Streams para supervisar las actividades de sus instancias de Amazon RDS. Con Database Activity Streams, Amazon RDS envía actividades al flujo de datos de Kinesis en tiempo real. A continuación, puede crear aplicaciones para la administración de la conformidad que consuman estas actividades, las auditen y generen alertas. También puede utilizar Amazon Data Firehose para almacenar los datos.
Para obtener más información, consulte [Database Activity Streams](https://docs.aws.amazon.com/AmazonRDS/latest/UserGuide/DBActivityStreams.html) en la *Guía para desarrolladores de Amazon RDS*.
# Escribir en Kinesis Data Streams mediante Amazon Pinpoint
Puede configurar Amazon Pinpoint para que envíe datos de eventos a Amazon Kinesis Data Streams. Amazon Pinpoint puede enviar datos de eventos para campañas, recorridos y mensajes transaccionales de email y SMS. A continuación, puede ingerir los datos en aplicaciones de análisis o crear sus propias aplicaciones de consumidor que realicen acciones basadas en el contenido de los eventos.
Para obtener más información, consulte [Transmisión de eventos](https://docs.aws.amazon.com/pinpoint/latest/developerguide/event-streams.html) en la *Guía para desarrolladores de Amazon Pinpoint*.
# Escribir en Kinesis Data Streams mediante Amazon Quantum Ledger Database (Amazon QLDB)
Puede crear un flujo en Amazon QLDB que capture todas las revisiones de documentos consignadas en su diario y entregue estos datos a Amazon Kinesis Data Streams en tiempo real. Un flujo de QLDB es un flujo continuo de datos desde el diario de su libro mayor a un recurso de flujo de datos de Kinesis. A continuación, puede utilizar la plataforma de streaming de Kinesis o Kinesis Client Library para consumir la transmisión, procesar los registros de datos y analizar el contenido de los datos. Un flujo de QLDB escribe sus datos en flujos de datos de Kinesis en tres tipos de registros: `control`, `block summary` y `revision details`.
Para obtener más información, consulte [Flujos](https://docs.aws.amazon.com/qldb/latest/developerguide/streams.html) en la *Guía para desarrolladores de Amazon QLDB*.
# Escribir en Kinesis Data Streams mediante integraciones de terceros
Puede escribir datos en Kinesis Data Streans mediante una de las siguientes opciones de terceros que se integran con Kinesis Data Streams. Seleccione la opción sobre la que desee obtener más información y busque recursos y enlaces a la documentación pertinente.
**Topics**
+ [Apache Flink](using-other-services-flink.md)
+ [Fluentd](using-other-services-Fluentd.md)
+ [Debezium](using-other-services-Debezium.md)
+ [Oráculo GoldenGate](using-other-services-Oracle-GoldenGate.md)
+ [Conexión de Kafka](using-other-services-kafka-connect.md)
+ [Adobe Experience](using-other-services-adobe.md)
+ [Striim](using-other-services-Striim.md)
# Apache Flink
Apache Flink es un marco y motor de procesamiento distribuido popular para computación con estado sobre flujos de datos ilimitados y delimitados. Para obtener más información sobre cómo escribir en Kinesis Data Streams desde Apache Flink, consulte [Amazon Kinesis Data Streams Connector](https://nightlies.apache.org/flink/flink-docs-release-1.15/docs/connectors/datastream/kinesis/).
# Fluentd
Fluentd es un recopilador de datos de código abierto para una capa de registro unificada. Para obtener más información sobre cómo escribir en Kinesis Data Streams desde Fluentd. Para obtener más información, consulte [Stream processing with Kinesis](https://docs.fluentd.org/how-to-guides/kinesis-stream).
# Debezium
Debezium es una plataforma distribuida de código abierto para la captura de datos sobre cambios. Para obtener más información sobre cómo escribir en Kinesis Data Streams desde Debezium, consulte [Streaming MySQL Data Changes to Amazon Kinesis](https://debezium.io/blog/2018/08/30/streaming-mysql-data-changes-into-kinesis/).
# Oráculo GoldenGate
Oracle GoldenGate es un producto de software que le permite replicar, filtrar y transformar datos de una base de datos a otra. Para obtener más información sobre cómo escribir en Kinesis Data Streams desde GoldenGate Oracle, [consulte Replicación de datos en Kinesis Data](https://blogs.oracle.com/dataintegration/post/data-replication-to-aws-kinesis-data-stream-using-oracle-goldengate) Stream mediante Oracle. GoldenGate
# Conexión de Kafka
Kafka Connect es una herramienta para transmitir datos de forma escalable y fiable entre Apache Kafka y otros sistemas. Para obtener más información sobre cómo escribir datos de Apache Kafka en Kinesis Data Streams, consulte [kinesis-kafka-connector](https://github.com/awslabs/kinesis-kafka-connector).
# Adobe Experience
Adobe Experience Platform permite a las organizaciones centralizar y estandarizar los datos de los clientes de cualquier sistema. Luego, aplica la ciencia de datos y el machine learning para mejorar drásticamente el diseño y la entrega de experiencias enriquecidas y personalizadas. Para obtener más información sobre cómo escribir datos de Adobe Experience Platform en Kinesis Data Streams, consulte cómo crear una [conexión de Amazon Kinesis](https://experienceleague.adobe.com/docs/experience-platform/destinations/catalog/cloud-storage/amazon-kinesis.html?lang=en).
# Striim
Striim es una plataforma completa integrada en memoria para recopilar, filtrar, transformar, enriquecer, agregar, analizar y entregar datos en tiempo real. end-to-end Para obtener más información sobre cómo escribir datos en Kinesis Data Streams desde Striim, consulte [Kinesis Writer](https://www.striim.com/docs/en/kinesis-writer.html).
# Solucionar problemas de los productores de Amazon Kinesis Data Streams
**Topics**
+ [Mi aplicación productora escribe a una velocidad menor que lo esperado](#producer-writing-at-slower-rate)
+ [Recibo un error de permiso no autorizado de clave maestra de KMS](#unauthorized-kms-producer)
+ [Solucionar otros problemas comunes para los productores](#misc-troubleshooting-producer)
## Mi aplicación productora escribe a una velocidad menor que lo esperado
**Topics**
+ [Límites de servicio superados](#service-limits-exceeded)
+ [Deseo optimizar mi productor](#producer-optimization)
+ [Uso indebido de las operaciones de `flushSync()`](#misuse-tag)
### Límites de servicio superados
Para saber si se superan los límites de servicio, compruebe si el productor crea excepciones de rendimiento desde el servicio y valide qué operaciones de la API se están viendo limitadas. Tenga en cuenta que hay diferentes límites en función de la llamada, consulte [Cuotas y límites](service-sizes-and-limits.md). Por ejemplo, además de los límites de nivel de fragmento para escrituras y lecturas más conocidos, existen los siguientes límites para el nivel de la secuencia:
+ [CreateStream](https://docs.aws.amazon.com/kinesis/latest/APIReference/API_CreateStream.html)
+ [DeleteStream](https://docs.aws.amazon.com/kinesis/latest/APIReference/API_DeleteStream.html)
+ [ListStreams](https://docs.aws.amazon.com/kinesis/latest/APIReference/API_ListStreams.html)
+ [GetShardIterator](https://docs.aws.amazon.com/kinesis/latest/APIReference/API_GetShardIterator.html)
+ [MergeShards](https://docs.aws.amazon.com/kinesis/latest/APIReference/API_MergeShards.html)
+ [DescribeStream](https://docs.aws.amazon.com/kinesis/latest/APIReference/API_DescribeStream.html)
+ [DescribeStreamSummary](https://docs.aws.amazon.com/kinesis/latest/APIReference/API_DescribeStreamSummary.html)
Las operaciones `CreateStream`, `DeleteStream`, `ListStreams`, `GetShardIterator` y `MergeShards` están limitadas a 5 llamadas por segundo. La operación `DescribeStream` está limitada a 10 llamadas por segundo. La operación `DescribeStreamSummary` está limitada a 20 llamadas por segundo.
Si estas llamadas no son el problema, asegúrese de que ha seleccionado una clave de partición que le permita distribuir de manera uniforme las operaciones *put* en todos los fragmentos, y que no tiene una clave de partición en particular que alcance los límites del servicio sin que el resto lo hagan. Esto requiere que mida los picos de rendimiento y que tenga en cuenta el número de fragmentos de su secuencia. Para obtener más información acerca de cómo administrar secuencias, consulte [Crear y administrar Kinesis Data Streams](working-with-streams.md).
**sugerencia**
Recuerde que debe redondear al kilobyte más cercano para calcular la limitación controlada del rendimiento al utilizar la operación de un solo registro [PutRecord](https://docs.aws.amazon.com/kinesis/latest/APIReference/API_PutRecord.html), mientras que la operación multirregistro [PutRecords](https://docs.aws.amazon.com/kinesis/latest/APIReference/API_PutRecords.html) redondea a la suma acumulativa de los registros en cada llamada. Por ejemplo, una solicitud `PutRecords` con 600 registros de 1,1 KB no será objeto de limitación controlada.
### Deseo optimizar mi productor
Antes de empezar a optimizar el productor, debe completar las siguientes tareas clave. En primer lugar, identifique el rendimiento máximo deseado en términos de tamaño del registro y registros por segundo. A continuación, descarte la capacidad de la secuencia como factor limitante ([Límites de servicio superados](#service-limits-exceeded)). Si ha descartado la capacidad de la secuencia, recurra a los siguientes consejos para solucionar problemas y directrices de optimización para los dos tipos comunes de productores.
**Productor grande**
Un productor grande se suele ejecutar desde un servidor en las instalaciones o una instancia de Amazon EC2. Los clientes que necesitan un mayor rendimiento de un productor grande normalmente se preocupan de la latencia por registro. Entre las estrategias para gestionar la latencia se incluyen las siguientes: si el cliente puede almacenar registros en microlotes o en búfer, utilice la [biblioteca Amazon Kinesis Producer](https://docs.aws.amazon.com/kinesis/latest/dev/developing-producers-with-kpl.html) (que cuenta con una lógica de agregación avanzada), la operación de varios registros o agregue los registros en un archivo más grande antes de utilizar la operación [PutRecords](https://docs.aws.amazon.com/kinesis/latest/APIReference/API_PutRecords.html)de registro único. [PutRecord](https://docs.aws.amazon.com/kinesis/latest/APIReference/API_PutRecord.html) Si no puede crear lotes o almacenar en memoria, use varios subprocesos para escribir en un servicio de Kinesis Data Streams al mismo tiempo. Esta AWS SDK para Java y otras SDKs incluyen clientes asíncronos que pueden hacerlo con muy poco código.
**Productores pequeños**
Un productor pequeño suele ser una aplicación móvil, un dispositivo de IoT o un cliente web. Si se trata de una aplicación móvil, le recomendamos que utilice la `PutRecords` operación o la grabadora Kinesis en el AWS dispositivo móvil. SDKs Para obtener más información, consulte la AWS Mobile SDK para Android Guía de introducción y la AWS Mobile SDK for iOS Guía de introducción. Las aplicaciones móviles deben administrar conexiones intermitentes inherentemente y necesitan algún tipo de inserción por lotes, como por ejemplo `PutRecords`. Si no puede crear lotes por algún motivo, consulte la información sobre productores grandes que aparece anteriormente. Si su productor es un navegador, la cantidad de datos generados suele ser muy pequeña. Sin embargo, estará ubicando las operaciones *put* en la ruta crítica de la aplicación, lo que no es recomendable.
### Uso indebido de las operaciones de `flushSync()`
Un uso incorrecto de `flushSync()` puede afectar notablemente al rendimiento de escritura. La operación `flushSync()` está diseñada para las situaciones de cierre, a fin de garantizar que todos los registros almacenados en búfer se envíen antes de que finalice la aplicación de KPL. Si se implementa esta operación después de cada operación de escritura, la latencia puede experimentar un aumento considerable de unos 500 ms por escritura. Asegúrese de implementar `flushSync()` únicamente para el cierre de la aplicación, a fin de evitar demoras adicionales innecesarias en el rendimiento de escritura.
## Recibo un error de permiso no autorizado de clave maestra de KMS
Este error se produce cuando una aplicación productora escribe en una secuencia cifrada sin permisos para la clave maestra de KMS. Para asignar permisos a una aplicación de modo que pueda obtener acceso a una clave de KMS, consulte [Uso de políticas de claves en AWS KMS](https://docs.aws.amazon.com/kms/latest/developerguide/key-policies.html) y [Uso de políticas de IAM con AWS KMS](https://docs.aws.amazon.com/kms/latest/developerguide/iam-policies.html).
## Solucionar otros problemas comunes para los productores
+ [¿Por qué mi flujo de datos de Kinesis devuelve un error de servidor interno 500?](https://aws.amazon.com/premiumsupport/knowledge-center/kinesis-data-stream-500-error/)
+ [¿Cómo soluciono los errores de tiempo de espera al escribir desde Flink a Kinesis Data Streams?](https://aws.amazon.com/premiumsupport/knowledge-center/kinesis-flink-timeout/)
+ [¿Cómo puedo solucionar los errores de limitación en Kinesis Data Streams?](https://aws.amazon.com/premiumsupport/knowledge-center/kinesis-data-stream-throttling-errors/)
+ [¿Por qué se limita mi flujo de datos de Kinesis?](https://aws.amazon.com/premiumsupport/knowledge-center/kinesis-data-stream-throttling/)
+ [¿Cómo puedo colocar registros de datos en un flujo de datos de Kinesis mediante KPL?](https://aws.amazon.com/premiumsupport/knowledge-center/kinesis-data-stream-kpl/)
# Optimizar sus productores de Amazon Kinesis Data Streams
Se puede optimizar incluso un poco más sus productores de Amazon Kinesis Data Streams según el comportamiento específico que ve. Revise los siguientes temas para identificar soluciones.
**Topics**
+ [Personalizar el comportamiento de reintentos y límites de velocidad en KPL.](kinesis-producer-adv-retries-rate-limiting.md)
+ [Aplicar las mejores prácticas a la agregación de KPL](kinesis-producer-adv-aggregation.md)
# Personalizar el comportamiento de reintentos y límites de velocidad en KPL.
Cuando agrega registros de usuario de Amazon Kinesis Producer Library (KPL) mediante la operación `addUserRecord()` de KPL, cada registro recibe una marca de tiempo y se agrega a un búfer con un plazo fijado por el parámetro de configuración `RecordMaxBufferedTime`. Esta stamp/deadline combinación de tiempo establece la prioridad del búfer. Los registros se vacían del búfer en función de los siguientes criterios:
+ Prioridad del búfer
+ Configuración de agregación
+ Configuración de recopilación
Los parámetros de configuración de agregación y recopilación que afectan al comportamiento del búfer son los siguientes:
+ `AggregationMaxCount`
+ `AggregationMaxSize`
+ `CollectionMaxCount`
+ `CollectionMaxSize`
A continuación, los registros vaciados se envían al flujo de datos de Kinesis como registros de Amazon Kinesis Data Streams a través de una llamada a la operación `PutRecords` de la API de Kinesis Data Streams. La operación `PutRecords` envía solicitudes a su secuencia que, de vez en cuando, presentan errores totales o parciales. Los registros con errores se agregan automáticamente al búfer de KPL. La nueva fecha límite se establece en función del mínimo de estos dos valores:
+ La mitad del valor configurado `RecordMaxBufferedTime` actual
+ El time-to-live valor del registro
Esta estrategia permite incluir los registros de usuarios de KPL reintentados en las siguientes llamadas a la API de Kinesis Data Streams para mejorar el rendimiento y reducir la complejidad y, al mismo tiempo, reforzar el valor del registro de Kinesis Data Streams. time-to-live No hay ningún algoritmo de retardo, por lo que esta estrategia de reintento resulta relativamente agresiva. El envío de spam debido a los reintentos excesivos se evita mediante una limitación de velocidad, que se aborda en la siguiente sección.
## Limitación de la velocidad
KPL incluye una característica de limitación de velocidad, que limita el rendimiento por partición enviada desde un único productor. La limitación de velocidad se implementa a través de un algoritmo de bucket de token con buckets independientes para registros y bytes de Kinesis Data Streams. Cada operación de escritura correcta en un flujo de datos de Kinesis agrega un token (o varios) a cada bucket, hasta un límite determinado. Este límite se puede configurar, pero de forma predeterminada se establece un 50 por ciento superior al límite real del fragmento, para permitir la saturación del fragmento desde un único productor.
Puede reducir este límite para disminuir el envío de spam debido a los reintentos excesivos. Sin embargo, es recomendable que cada productor intente por todos los medios alcanzar el máximo rendimiento y que administre las limitaciones controladas resultantes determinadas como excesivas mediante la expansión de la capacidad de la secuencia y la implementación de una estrategia adecuada para la clave de partición.
# Aplicar las mejores prácticas a la agregación de KPL
Aunque el esquema de números secuenciales de los registros de Amazon Kinesis Data Streams resultantes sigue siendo el mismo, la agregación hace que la indexación de registros de usuario de Amazon Kinesis Producer Library (KPL) contenidos en un registro agregado de Kinesis Data Streams comience en 0 (cero); sin embargo, siempre y cuando no dependa de los números secuenciales para identificar exclusivamente sus registros de usuario de KPL, el código puede pasar por alto esta información, ya que la agregación (de los registros de usuario de KPL en un registro de Kinesis Data Streams) y la desagrupación posterior (del registro de Kinesis Data Streams en registros de usuario de KPL) se encarga de todo eso automáticamente. Esto se aplica tanto si su consumidor utiliza la KCL como el SDK. AWS Para usar esta funcionalidad de agregación, tendrás que incorporar la parte de Java de la KPL a tu compilación si tu usuario está escrito con la API proporcionada en el AWS SDK.
Si va a utilizar números secuenciales como identificadores únicos para sus registros de usuario de KPL, se recomienda que utilice las operaciones persistentes `public int hashCode()` y `public boolean equals(Object obj)` proporcionadas en `Record` y `UserRecord` para permitir la comparación de sus registros de usuario de KPL. Además, si desea examinar el número subsecuencial de su registro de usuario de KPL, puede transmitirlo a una instancia `UserRecord` y recuperar su número subsecuencial.
Para obtener más información, consulte [Implementar la desagregación de consumidores](kinesis-kpl-consumer-deaggregation.md).