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. # Leer datos de Amazon Kinesis Data Streams Un *consumidor* es una aplicación que procesa todos los datos procedentes de un flujo de datos de Kinesis. Cuando un consumidor utiliza una *distribución mejorada*, obtiene sus propias 2 MB/sec porciones de rendimiento de lectura, lo que permite a varios consumidores leer datos de la misma secuencia en paralelo, sin competir por el rendimiento de lectura con otros consumidores. Para utilizar la característica de distribución ramificada mejorada de los fragmentos, consulte [Desarrollo de consumidores de distribución ramificada mejorada con rendimiento dedicado](enhanced-consumers.md). Puede crear consumidores para Kinesis Data Streams mediante Kinesis Client Library (KCL) o AWS SDK para Java. También puede aumentar el número de consumidores mediante otros AWS servicios AWS Lambda, como Amazon Managed Service para Apache Flink y Amazon Data Firehose. Kinesis Data Streams admite integraciones con AWS otros servicios, como Amazon EMR, Amazon AWS Glue y EventBridge Amazon Redshift. También es compatible con integraciones de terceros, como Apache Flink, Adobe Experience Platform, Apache Druid, Apache Spark, Databricks, Confluent Platform, Kinesumer y Talend. **Topics** + [ # Desarrollo de consumidores de distribución ramificada mejorada con rendimiento dedicado ](enhanced-consumers.md) + [ # Usar el visor de datos en la consola de Kinesis ](data-viewer.md) + [ # Consultar los flujos de datos en la consola de Kinesis ](querying-data.md) + [ # Uso de Kinesis Client Library ](kcl.md) + [ # Desarrolle a los consumidores con AWS SDK para Java ](develop-consumers-sdk.md) + [ # Desarrolle a los consumidores utilizando AWS Lambda ](lambda-consumer.md) + [ # Desarrollar consumidores mediante Amazon Managed Service para Apache Flink ](kda-consumer.md) + [ # Desarrollar consumidores mediante Amazon Data Firehose ](kdf-consumer.md) + [ # Lea datos de Kinesis Data Streams mediante otros servicios AWS ](using-other-services-read.md) + [ # Leer los datos de Kinesis Data Streams mediante integraciones de terceros ](using-services-third-party-read.md) + [ # Solución de problemas de los consumidores de Kinesis Data Streams ](troubleshooting-consumers.md) + [ # Optimizar consumidores de Amazon Kinesis Data Streams. ](advanced-consumers.md) # Desarrollo de consumidores de distribución ramificada mejorada con rendimiento dedicado En Amazon Kinesis Data Streams puede crear consumidores que utilicen una característica denominada *distribución ramificada mejorada*. Esta característica permite a los consumidores recibir registros de un flujo con un rendimiento de hasta 2 MB de datos por segundo por partición. Este rendimiento está dedicado, lo que significa que los consumidores que utilizan la distribución ramificada mejorada no tienen que competir con otros consumidores que reciben datos del flujo. Kinesis Data Streams inserta registros de datos desde el flujo en los consumidores que utilizan la distribución ramificada mejorada. Por lo tanto, estos consumidores no necesitan sondear los datos. **importante** Con el modo On-Demand Advantage, puedes registrar hasta 50 consumidores por transmisión para utilizar una distribución mejorada. Con las transmisiones bajo demanda estándar y aprovisionadas, puedes registrar hasta 20 consumidores por transmisión para utilizar la distribución mejorada. En el siguiente diagrama, se muestra la arquitectura de distribución ramificada mejorada. Si utiliza la versión 2.0 o posterior de Amazon Kinesis Client Library (KCL) para crear un consumidor, KCL configura el consumidor para que utilice la distribución ramificada mejorada para recibir datos de todas las particiones del flujo. Si utiliza la API para crear un consumidor que utiliza la distribución ramificada mejorada, puede suscribirse a fragmentos individuales. ![\[Diagrama de flujo que muestra la arquitectura de distribución ramificada mejorada con dos fragmentos y dos consumidores. Cada uno de los dos consumidores utiliza la distribución ramificada mejorada para recibir datos de ambos fragmentos de la secuencia.\]](http://docs.aws.amazon.com/es_es/streams/latest/dev/images/enhanced_fan-out.png) En el diagrama se muestra lo siguiente: + Una secuencia con dos fragmentos. + Dos consumidores que utilizan la distribución ramificada mejorada para recibir datos de la secuencia: Consumer X y Consumer Y. Ambos están suscritos a todos los fragmentos y a todos los registros de la secuencia. Si utiliza la versión 2.0 o posterior de KCL para crear un consumidor, KCL suscribe automáticamente ese consumidor a todas las particiones del flujo. Por otro lado, si utiliza la API para crear un consumidor, puede suscribirse a fragmentos individuales. + Flechas que representan las canalizaciones de distribución ramificada mejorada que utilizan los consumidores para recibir datos de la secuencia. Una canalización de distribución mejorada proporciona hasta un 2% MB/sec de datos por fragmento, independientemente de cualquier otra canalización o del número total de consumidores. **Topics** + [ ## Diferencias entre el consumidor de rendimiento compartido y el consumidor de distribución ramificada mejorada ](#enhanced-consumers-differences) + [ ## Compatible con regiones para un máximo de 50 usuarios con sistema de distribución mejorada (solo On-Demand Advantage) ](#supported-regions) + [ # Gestione la mayor distribución de los consumidores con el o AWS CLI APIs ](building-enhanced-consumers-console.md) ## Diferencias entre el consumidor de rendimiento compartido y el consumidor de distribución ramificada mejorada En la tabla siguiente, se comparan los consumidores de rendimiento compartido predeterminado y los consumidores de distribución ramificada mejorada. El retardo de propagación del mensaje se define como el tiempo en milisegundos que tarda una carga útil enviada mediante el envío de la carga útil APIs (como `PutRecord` y`PutRecords`) en llegar a la aplicación de consumo a través de la carga útil que consume (como y). APIs `GetRecords` `SubscribeToShard` **En esta tabla se comparan los consumidores de rendimiento compartido y los consumidores de distribución ramificada mejorada** | Características | Consumidores de rendimiento compartido sin una distribución ramificada mejorada | Consumidores de distribución ramificada mejorada | | --- | --- | --- | | Velocidad de lectura | Se ha fijado en un total de 2 por fragmento. MB/sec Si hay varios consumidores que leen en un mismo fragmento, todos ellos comparten este rendimiento. La suma de rendimientos que reciben desde el fragmento no supera los 2 MB/s. | Se escala a medida que los consumidores se registran para utilizar la distribución ramificada mejorada. Cada consumidor registrado para usar un despliegue mejorado recibe su propio rendimiento de lectura por fragmento, hasta 2 MB/seg, independientemente de otros consumidores. | | Retraso de propagación de mensajes | Un promedio de alrededor de 200 ms si hay un consumidor leyendo la secuencia. Este promedio alcanza aproximadamente los 1000 ms si hay cinco consumidores. | Por lo general, el promedio es de 70 ms si hay un consumidor o cinco consumidores. | | Costo | No aplicable | Existe un costo de recuperación de datos y un costo por hora y fragmento para los consumidores. Para obtener más información, consulte los [precios de Amazon Kinesis Data Streams](https://aws.amazon.com/kinesis/data-streams/pricing/?nc=sn&loc=3). | | Modelo de entrega de registros | Utilice HTTP para colocar el modelo en lugar de HTTP. GetRecords | Kinesis Data Streams le envía los registros a través de HTTP/2 mediante. SubscribeToShard | ## Compatible con regiones para un máximo de 50 usuarios con sistema de distribución mejorada (solo On-Demand Advantage) El soporte para hasta 50 consumidores con distribución mejorada en el modo On-Demand Advantage solo está disponible en las siguientes AWS regiones: | AWS Región | Nombre de la región | | --- | --- | | eu-north-1 | Europa (Estocolmo) | | me-south-1 | Middle East (Bahrain) | | ap-south-1 | Asia-Pacífico (Mumbai) | | eu-west-3 | Europa (París) | | ap-southeast-3 | Asia-Pacífico (Yakarta) | | us-east-2 | Este de EE. UU. (Ohio) | | af-south-1 | África (Ciudad del Cabo) | | eu-west-1 | Europa (Irlanda) | | me-central-1 | Medio Oriente (EAU) | | eu-central-1 | Europa (Fráncfort) | | sa-east-1 | América del Sur (São Paulo) | | ap-east-1 | Asia-Pacífico (Hong Kong) | | ap-south-2 | Asia-Pacífico (Hyderabad) | | us-east-1 | Este de EE. UU. (Norte de Virginia) | | ap-northeast-2 | Asia-Pacífico (Seúl) | | ap-northeast-3 | Asia-Pacífico (Osaka) | | eu-west-2 | Europa (Londres) | | ap-southeast-4 | Asia-Pacífico (Melbourne) | | ap-northeast-1 | Asia-Pacífico (Tokio) | | us-west-2 | Oeste de EE. UU. (Oregón) | | us-west-1 | Oeste de EE. UU. (Norte de California) | | ap-southeast-1 | Asia-Pacífico (Singapur) | | ap-southeast-2 | Asia-Pacífico (Sídney) | | il-central-1 | Israel (Tel Aviv) | | ca-central-1 | Canadá (centro) | | ca-west-1 | Oeste de Canadá (Calgary) | | eu-south-2 | Europa (España) | | cn-northwest-1 | China (Ningxia) | | eu-central-2 | Europa (Zúrich) | | us-gov-east-1 | AWS GovCloud (Este de EE. UU.) | | us-gov-west-1 | AWS GovCloud (EE. UU.-Oeste) | # Gestione la mayor distribución de los consumidores con el o AWS CLI APIs Los consumidores que utilizan la *distribución ramificada mejorada* en Amazon Kinesis Data Streams pueden recibir los registros de un flujo de datos con un rendimiento dedicado de hasta 2 MB de datos por segundo por partición. Para obtener más información, consulte [Desarrollo de consumidores de distribución ramificada mejorada con rendimiento dedicado](enhanced-consumers.md). Puede usar AWS CLI Kinesis Data APIs Streams para registrar, describir, enumerar y anular el registro de un consumidor que utilice la distribución mejorada en Kinesis Data Streams. ## Administre a los consumidores mediante AWS CLI Puede registrar, describir, enumerar y anular el registro de los consumidores con mayor número de seguidores utilizando el. AWS CLI Para ver ejemplos, consulte la documentación siguiente. [register-stream-consumer](https://awscli.amazonaws.com/v2/documentation/api/latest/reference/kinesis/register-stream-consumer.html) Registra un consumidor para un flujo de datos de Kinesis. Puede aplicar etiquetas al registrar al consumidor. [describe-stream-consumer](https://awscli.amazonaws.com/v2/documentation/api/latest/reference/kinesis/describe-stream-consumer.html) Obtiene la descripción de un consumidor registrado con un ARN de consumidor o una combinación de nombre de consumidor y ARN de flujo. [list-stream-consumers](https://awscli.amazonaws.com/v2/documentation/api/latest/reference/kinesis/list-stream-consumers.html) Muestra una lista de consumidores registrados para recibir datos de un flujo mediante una distribución ramificada mejorada. [deregister-stream-consumer](https://awscli.amazonaws.com/v2/documentation/api/latest/reference/kinesis/deregister-stream-consumer.html) Anule el registro de un consumidor con una combinación de ARN de consumidor o nombre de consumidor y ARN de flujo. ## Administre a los consumidores mediante Kinesis Data Streams APIs Puede registrar, describir, enumerar y anular el registro de los consumidores con una distribución mejorada mediante Kinesis Data Streams. APIs Para ver ejemplos, consulte la documentación siguiente. [RegisterStreamConsumer](https://docs.aws.amazon.com/kinesis/latest/APIReference/API_RegisterStreamConsumer.html) Registra un consumidor para un flujo de datos de Kinesis con etiquetas. Puede aplicar etiquetas al registrar al consumidor. [DescribeStreamConsumer](https://docs.aws.amazon.com/kinesis/latest/APIReference/API_DescribeStreamConsumer.html) Obtiene la descripción de un consumidor registrado con un ARN de consumidor o una combinación de nombre de consumidor y ARN de flujo. [ListStreamConsumers](https://docs.aws.amazon.com/kinesis/latest/APIReference/API_ListStreamConsumers.html) Muestra una lista de consumidores registrados para recibir datos de un flujo mediante una distribución ramificada mejorada. [DeregisterStreamConsumer](https://docs.aws.amazon.com/kinesis/latest/APIReference/API_DeregisterStreamConsumer.html) Anule el registro de un consumidor con una combinación de ARN de consumidor o nombre de consumidor y ARN de flujo. ## Etiquetar consumidores Puede asignar sus propios metadatos a los flujos y a los consumidores de distribución ramificada mejorada que cree en Kinesis Data Streams en forma de etiquetas. Puede utilizar etiquetas para categorizar y hacer un seguimiento de los costos de sus consumidores. También puede controlar el acceso a los consumidores mediante etiquetas con el [control de acceso basado en atributos (ABAC)](https://docs.aws.amazon.com/IAM/latest/UserGuide/introduction_attribute-based-access-control.html). Para obtener más información, consulte [Etiquetar recursos de Amazon Kinesis Data Streams](tagging.md). # Usar el visor de datos en la consola de Kinesis El visor de datos de la consola de administración de Kinesis le permite ver los registros de datos dentro de la partición especificada de su flujo de datos sin tener que desarrollar una aplicación de consumo. Para utilizar el visor de datos, siga estos pasos: 1. [Inicie sesión en la consola de Kinesis Consola de administración de AWS y ábrala en https://console.aws.amazon.com /kinesis.](https://console.aws.amazon.com/kinesis) 1. Elija el flujo de datos activo cuyos registros desee ver con el visor de datos y, a continuación, elija la pestaña **Visor de datos**. 1. En la pestaña **Visor de datos** del flujo de datos activo seleccionado, elija la partición cuyos registros desee ver, elija la **posición inicial** y, a continuación, haga clic en **Obtener registros**. Puede establecer la posición de inicio en uno de los siguientes valores: + **En el número secuencial**: muestra los registros desde la posición indicada por el número secuencial especificado en el campo de números secuenciales. + **Después del número secuencial**: muestra los registros después de la posición indicada por el número secuencial especificado en el campo de números secuenciales. + **En la marca de tiempo**: muestra los registros desde la posición indicada por la marca de tiempo especificada en el campo de marca de tiempo. + **Recortar horizonte**: muestra los registros en el último registro sin recortar de la partición, que es el registro de datos más antiguo de esta. + **Más reciente**: muestra los registros justo después del registro más reciente de la partición, de modo que siempre se lean los datos más recientes de la partición. Los registros de datos generados que coinciden con el ID de partición y la posición inicial especificados se muestran entonces en una tabla de registros en la consola. Se muestran un máximo de 50 registros a la vez. Para ver el siguiente conjunto de registros, haga clic en el botón **Siguiente**. 1. Haga clic en cualquier registro individual para ver la carga de ese registro en datos sin procesar o en formato JSON en una ventana independiente. Tenga en cuenta que al hacer clic en los botones **Obtener registros** o **Siguiente** del **Visor de datos**, se invoca la **GetRecords**API y se aplica al límite de la **GetRecords**API de 5 transacciones por segundo. # Consultar los flujos de datos en la consola de Kinesis La pestaña Análisis de datos de la consola de Kinesis Data Streams le permite consultar sus flujos de datos mediante SQL. Para utilizar esta capacidad, siga estos pasos: 1. [Inicie sesión en la consola de Kinesis Consola de administración de AWS y ábrala en https://console.aws.amazon.com /kinesis.](https://console.aws.amazon.com/kinesis) 1. Elija el flujo de datos activo que desee consultar con SQL y, a continuación, elija la pestaña **Análisis de datos**. 1. En la pestaña **Análisis de datos**, puede inspeccionar y visualizar el flujo con un bloc de notas Managed Apache Flink Studio. Puede realizar consultas SQL ad hoc para inspeccionar su flujo de datos y ver los resultados en cuestión de segundos con Apache Zeppelin. En la pestaña **Análisis de datos**, seleccione **Estoy de acuerdo** y, a continuación, seleccione **Crear bloc de notas** para crear un bloc de notas. 1. Una vez creado el bloc de notas, seleccione **Abrir en Apache Zeppelin**. De este modo, se abre el bloc de notas en una nueva pestaña. Un cuaderno es una interfaz interactiva en la que puede enviar sus consultas SQL. Elija la nota que contiene el nombre de su flujo. 1. Verá una nota con un ejemplo de consulta `SELECT` para mostrar los datos del flujo que ya se está ejecutando. Esto le permite ver el esquema de su flujo de datos. 1. Para probar otras consultas, como ventanas plegables o deslizantes, seleccione **Ver consultas de muestra** en la pestaña **Análisis de datos**. Copie la consulta, modifíquela para adaptarla a su esquema de flujo de datos y, a continuación, ejecútela en un nuevo párrafo de su nota de Zeppelin. # Uso de Kinesis Client Library ## ¿Qué es Kinesis Client Library? Kinesis Client Library (KCL) es una biblioteca de software Java independiente diseñada para facilitar el proceso de consumo y procesamiento de datos de Amazon Kinesis Data Streams. KCL gestiona numerosas tareas complejas que están asociadas a la computación distribuida, lo que le permite a los desarrolladores implementar su lógica empresarial para procesar datos. Administra actividades como equilibrar la carga entre varios procesos de trabajo, responder a los fallos de los procesos de trabajo, integrar puntos de control de los registros procesados y responder a los cambios en la cantidad de particiones en el flujo. KCL se actualiza con frecuencia para incorporar versiones más recientes de las bibliotecas subyacentes, mejoras de seguridad y correcciones de errores. Recomendamos utilizar la versión más reciente de KCL para evitar los problemas ya conocidos y beneficiarse de las mejoras actuales. Para encontrar la versión más reciente de KCL, consulte [KCL GitHub](https://github.com/awslabs/amazon-kinesis-client). **importante** Recomendamos utilizar la versión más reciente de KCL para evitar errores y problemas ya conocidos. Si utiliza KCL 2.6.0 o una versión anterior, actualice a KCL 2.6.1, o a una versión posterior, para evitar bloquear el procesamiento de las particiones cuando varía la capacidad del flujo (caso poco común). KCL es una biblioteca de Java. Support para lenguajes distintos de Java se proporciona mediante un daemon basado en Java llamado. MultiLangDaemon MultiLangDaemoninteractúa con la aplicación KCL a través de STDIN y STDOUT. Para obtener más información sobre el encendido, consulte. MultiLangDaemon GitHub [Desarrollar consumidores con KCL en lenguajes distintos de Java](develop-kcl-consumers-non-java.md) No utilice las AWS SDK para Java versiones 2.27.19 a 2.27.23 con KCL 3.x. Estas versiones cuentan con un problema que provoca un error de excepción relacionado con el uso de DynamoDB por parte de KCL. Le recomendamos que utilice la AWS SDK para Java versión 2.28.0 o posterior para evitar este problema. ## Características y ventajas clave de KCL A continuación se indican las características claves y los beneficios relacionados de KCL: + **Escalabilidad**: KCL permite que las aplicaciones se escalen de forma dinámica al distribuir la carga de procesamiento entre varios procesos de trabajo. Puede escalar su aplicación hacia dentro o hacia fuera, de forma manual o mediante escalado automático, sin preocuparse por la redistribución de la carga. + **Equilibrador de carga**: KCL equilibra automáticamente la carga de procesamiento entre los procesos de trabajo disponibles, lo que resulta en una distribución uniforme del trabajo entre ellos. + **Puntos de control**: KCL administra el registro de los puntos de control de los registros procesados, lo que permite a las aplicaciones reanudar el procesamiento desde la última posición en la que se procesaron correctamente. + **Tolerancia a errores**: KCL proporciona mecanismos de tolerancia a errores integrados, lo que garantiza que el procesamiento de datos continúe incluso si los procesos de trabajo fallan. KCL también proporciona servicios de entrega. at-least-once + **Gestión de los cambios a nivel de flujo**: KCL se adapta a las divisiones y fusiones de particiones que pueden producirse debido a cambios en el volumen de datos. Mantiene el orden asegurándose de que las particiones secundarias se procesen solo después de que la partición principal se haya completado y registrado su punto de control. + **Supervisión**: KCL se integra con Amazon CloudWatch para la supervisión a nivel de consumidor. + **Soporte multilingüe**: KCL es compatible de forma nativa con Java y permite utilizar varios lenguajes de programación distintos de Java. MultiLangDaemon # Conceptos de KCL En esta sección se explican los conceptos básicos y las interacciones de Kinesis Client Library (KCL). Estos conceptos son esenciales para desarrollar y administrar las aplicaciones de consumo de KCL. + **Aplicación de consumo de KCL**: una aplicación creada a medida y diseñada para leer y procesar registros de flujos de datos de Kinesis mediante Kinesis Client Library. + **Proceso de trabajo**: las aplicaciones de consumo de KCL suelen estar distribuidas, y uno o más procesos de trabajo se ejecutan simultáneamente. KCL coordina los procesos de trabajo para que consuman los datos del flujo de manera distribuida y equilibra la carga de manera uniforme entre varios procesos de trabajo. + **Programador**: clase de alto nivel que utiliza un proceso de trabajo de KCL para empezar a procesar datos. Cada proceso de trabajo de KCL tiene un programador. El programador inicializa y supervisa diversas tareas, como la sincronización de la información sobre las particiones de los flujos de datos de Kinesis, el seguimiento de las asignaciones de las particiones entre los procesos de trabajo y el procesamiento de los datos desde el flujo según las particiones asignadas al proceso de trabajo. El programador puede adoptar diversas configuraciones que afectan al comportamiento de este, como el nombre del flujo que se va a procesar y las credenciales de AWS , El programador inicia la entrega de los registros de datos del flujo a los procesadores de registros. + **Procesador de registros**: define la lógica de la forma en que su aplicación de consumo de KCL procesa los datos que obtiene de los flujos de datos. Debe implementar su propia lógica de procesamiento de datos personalizada en el procesador de registros. Un proceso de trabajo de KCL crea una instancia de un programador. Luego, el programador crea una instancia de un procesador de registros por cada partición que mantiene un arrendamiento. Un proceso de trabajo puede ejecutar varios procesadores de registros. + **Arrendamiento**: define la asignación entre un proceso de trabajo y una partición. Las aplicaciones de consumo de KCL utilizan los arrendamientos para dividir el procesamiento de registro de datos entre varios procesos de trabajo. Cada partición está vinculada a un solo proceso de trabajo mediante un arrendamiento en un momento dado y cada proceso de trabajo puede mantener uno o más arrendamientos simultáneamente. Cuando un proceso de trabajo deja de mantener un arrendamiento debido a que se detuvo o ocurrió un error, KCL asigna a otro para que se quede con el arrendamiento. Para obtener más información sobre el arrendamiento, consulte la [Documentación de Github: Lease Lifecycle](https://github.com/awslabs/amazon-kinesis-client/blob/master/docs/lease-lifecycle.md#lease-lifecycle). + **Tabla de arrendamiento**: es una tabla exclusiva de Amazon DynamoDB que se utiliza para realizar un seguimiento de todos los arrendamientos de la aplicación de consumo de KCL. Cada aplicación de consumo de KCL crea su propia tabla de arrendamiento. La tabla de arrendamiento se utiliza para mantener el estado de todos los procesos de trabajo y coordinar el procesamiento de los datos. Para obtener más información, consulte [Tablas de metadatos de DynamoDB y equilibrio de carga en KCL](kcl-dynamoDB.md). + **Registro de puntos de control**: es el proceso de almacenar de forma persistente en una partición la posición del último registro procesado correctamente. KCL administra el registro de puntos de control para garantizar que el procesamiento se pueda reanudar desde la última posición del punto de control en caso de que un proceso de trabajo falle o se reinicie la aplicación. Los puntos de control se almacenan en la tabla de arrendamiento de DynamoDB como parte de los metadatos del arrendamiento. Esto permite a los procesos de trabajo continuar procesando desde donde se detuvo el proceso de trabajo anterior. # Tablas de metadatos de DynamoDB y equilibrio de carga en KCL KCL administra los metadatos, como los arrendamientos y las métricas de uso de la CPU de los procesos de trabajo. KCL rastrea estos metadatos mediante tablas de DynamoDB. Para cada aplicación de Amazon Kinesis Data Streams, KCL crea tres tablas de DynamoDB para administrar los metadatos: la tabla de arrendamiento, la tabla de métricas de procesos de trabajo y la de estado de coordinadores. **nota** KCL 3.x presentó dos nuevas tablas de metadatos: las tablas de *métricas de procesos de trabajo* y de *estado de coordinadores*. **importante** Debe agregar los permisos adecuados para que las aplicaciones de KCL puedan crear y administrar tablas de metadatos en DynamoDB. Para obtener más información, consulte [Permisos de IAM necesarios para las aplicaciones de consumo de KCL](kcl-iam-permissions.md). La aplicación de consumo KCL no elimina automáticamente estas tres tablas de metadatos de DynamoDB. Asegúrese de eliminar estas tablas de metadatos creadas por la aplicación de consumo de KCL al retirar la aplicación de consumo para evitar costos innecesarios. ## Tabla de arrendamiento Una tabla de arrendamiento es una tabla exclusiva de Amazon DynamoDB que se utiliza para realizar un seguimiento de las particiones que los programadores de la aplicación de consumo de KCL están arrendando y procesando. Cada aplicación de consumo de KCL crea su propia tabla de arrendamiento. La KCL, por defecto, utiliza el nombre de la aplicación de consumo para el nombre de la tabla de arrendamiento. Puede definir un nombre de tabla personalizado con la configuración. KCL también crea un [índice secundario global](https://docs.aws.amazon.com/amazondynamodb/latest/developerguide/GSI.html) en la tabla de arrendamiento con la clave de partición de leaseOwner para detectar los arrendamientos de manera eficiente. El índice secundario global refleja el atributo leaseKey de la tabla de arrendamientos base. Si la tabla de arrendamiento de la aplicación de consumo de KCL no existe cuando se inicia la aplicación, uno de los procesos de trabajo creará la tabla de arrendamiento para su aplicación. Puede consultar la tabla con la [consola de Amazon DynamoDB](https://docs.aws.amazon.com/amazondynamodb/latest/developerguide/ConsoleDynamoDB.html) mientras se ejecuta la aplicación de consumo. **importante** El nombre de cada aplicación de consumo de KCL debe ser único para evitar que se duplique el nombre de la tabla de arrendamiento. Se le realizará el cobro de los costos de su cuenta asociados a la tabla de DynamoDB, además de los costos propios asociados a Kinesis Data Streams. Cada fila de la tabla de arrendamiento representa una partición que procesan los programadores de su aplicación de consumo. Algunos de los campos claves son los siguientes: + **leaseKey**: este es el ID de partición para el procesamiento de un solo flujo. Para el procesamiento de varios flujos con KCL, se estructura como `account-id:StreamName:streamCreationTimestamp:ShardId`. leaseKey es la clave de partición de la tabla de arrendamiento. Para obtener más información sobre el procesamiento de varios flujos, consulte [Procesamiento de varios flujos con KCL](kcl-multi-stream.md). + **checkpoint:** el número secuencial de punto de comprobación más reciente del fragmento. + **checkpointSubSequenceNúmero:** cuando se utiliza la función de agregación de la biblioteca de productores de Kinesis, se trata de una extensión del **punto de control** que rastrea los registros de los usuarios individuales dentro del registro de Kinesis. + **leaseCounter:** se utiliza para comprobar si un proceso de trabajo está procesando el arrendamiento de forma activa. leaseCounter aumenta si la propiedad del arrendamiento se transfiere a otro proceso de trabajo. + **leaseOwner:** el proceso de trabajo actual que mantiene este arrendamiento. + **ownerSwitchesSincePunto de control:** cuántas veces en este contrato de arrendamiento se han cambiado de trabajadores desde el último punto de control. + **parentShardId:** ID del padre de este fragmento. Se asegura de que la partición origen esté completamente procesada antes de que comience el procesamiento en las particiones secundarias, lo que mantiene el orden de procesamiento de registros correcto. + **childShardId:** Lista de fragmentos secundarios IDs resultantes de la división o fusión de este fragmento. Se utiliza para rastrear el linaje de la partición y administrar el orden de procesamiento durante las operaciones de repartición. + **startingHashKey:** El límite inferior del rango de claves hash de este fragmento. + **endingHashKey:** El límite superior del rango de claves de hash de este fragmento. Si utiliza el procesamiento de varios flujos con KCL, verá los dos siguientes campos adicionales en la tabla de arrendamiento. Para obtener más información, consulte [Procesamiento de varios flujos con KCL](kcl-multi-stream.md). + **shardID:** ID de la partición. + **streamName:** identificador del flujo de datos en el siguiente formato: `account-id:StreamName:streamCreationTimestamp`. ## Tabla de métricas de procesos de trabajo La tabla de métricas de procesos de trabajo es una tabla única de Amazon DynamoDB para cada aplicación de KCL que se utiliza para registrar las métricas de uso de la CPU en cada proceso de trabajo. KCL utilizará estas métricas para realizar asignaciones de arrendamiento eficientes a fin de lograr un uso equilibrado de los recursos entre los procesos de trabajo. Por defecto, KCL utiliza `KCLApplicationName-WorkerMetricStats` para el nombre de la tabla de métricas de procesos de trabajo. ## Tabla de estado de coordinadores Una tabla de estado de coordinadores es una tabla de Amazon DynamoDB exclusiva para cada aplicación de KCL y se utiliza para almacenar información de estado interna de los procesos de trabajo. Por ejemplo, la tabla de estado de coordinadores almacena datos relacionados con la elección del líder o los metadatos asociados a la migración local de KCL 2.x a KCL 3.x. Por defecto, KCL utiliza `KCLApplicationName-CoordinatorState` como el nombre de la tabla de estado de coordinadores. ## Modo de capacidad de DynamoDB para las tablas de metadatos creadas por KCL Por defecto, Kinesis Client Library (KCL) crea tablas de metadatos de DynamoDB, como la tabla de arrendamiento, la tabla de métricas de procesos de trabajo y la tabla de estado de coordinadores, mediante el [modo de capacidad bajo demanda](https://docs.aws.amazon.com/amazondynamodb/latest/developerguide/on-demand-capacity-mode.html). Este modo escala la capacidad de lectura y escritura de forma automática para adaptarse al tráfico sin necesidad de planificar la capacidad. Recomendamos mantener el modo de capacidad como bajo demanda para un funcionamiento más eficiente de estas tablas de metadatos. Si decide cambiar la tabla de arrendamiento al [modo de capacidad aprovisionada](https://docs.aws.amazon.com/amazondynamodb/latest/developerguide/provisioned-capacity-mode.html), siga estas prácticas recomendadas: + Analice los patrones de uso: + Supervisa los patrones y usos de lectura y escritura de tu aplicación (RCU, WCU) mediante las métricas de Amazon. CloudWatch + Conozca los requisitos de rendimiento máximo y medio. + Calcule la capacidad necesaria: + Calcule las unidades de capacidad de lectura (RCUs) y las unidades de capacidad de escritura (WCUs) en función de su análisis. + Contemple factores como la cantidad de particiones, la frecuencia de los puntos de control y la cantidad de procesos de trabajo. + Implemente el escalado automático: + Utilice el [escalado automático de DynamoDB](https://docs.aws.amazon.com/amazondynamodb/latest/developerguide/provisioned-capacity-mode.html#ddb-autoscaling) para ajustar automáticamente la capacidad aprovisionada y establecer los límites de capacidad mínima y máxima adecuados. + El escalado automático de DynamoDB ayudará a evitar que la tabla de metadatos de KCL alcance el límite de capacidad y se vea limitada. + Supervise y optimice periódicamente: + Supervise continuamente CloudWatch las métricas para`ThrottledRequests`. + Ajuste la capacidad a medida que su carga de trabajo cambie con el tiempo. Si se encuentra con `ProvisionedThroughputExceededException` en las tablas de metadatos de DynamoDB de su aplicación de consumo de KCL, deberá aumentar la capacidad de rendimiento aprovisionada de la tabla de DynamoDB. Si establece un nivel determinado de unidades de capacidad de lectura (RCU) y unidades de capacidad de escritura (WCU) al crear la aplicación de consumo por primera vez, puede que no sea suficiente a medida que aumente su uso. Por ejemplo, si su aplicación de consumo de KCL realiza registros de puntos de control frecuentes u opera en un flujo con muchas particiones, es posible que necesite más unidades de capacidad. Para obtener información sobre el rendimiento aprovisionado en DynamoDB, consulte [Capacidad de rendimiento de DynamoDB](https://docs.aws.amazon.com/amazondynamodb/latest/developerguide/capacity-mode.html) y [actualización de una tabla](https://docs.aws.amazon.com/amazondynamodb/latest/developerguide/WorkingWithTables.Basics.html#WorkingWithTables.Basics.UpdateTable) en la Guía para desarrolladores de Amazon DynamoDB. ## Cómo KCL asigna los arrendamientos a los procesos de trabajo y equilibra la carga KCL recopila y supervisa continuamente las métricas de uso de la CPU de los hosts de procesamiento que ejecutan los procesos de trabajo para garantizar una distribución uniforme de la carga de trabajo. Estas métricas de uso de la CPU se almacenan en la tabla de métricas de procesos de trabajo de DynamoDB. Si KCL detecta que algunos procesos de trabajo muestran índices de uso de la CPU más altos que otros, reasignará los arrendamientos entre ellos para reducir la alta carga de trabajo. El objetivo es equilibrar la carga de trabajo de manera más uniforme en toda la flota de aplicaciones de consumo, evitando que un solo proceso de trabajo se sobrecargue. A medida que KCL distribuye la utilización de la CPU entre la flota de aplicaciones de consumo, puede ajustar la capacidad de su flota de aplicaciones de consumo al elegir el número correcto de procesos de trabajo o al utilizar el escalado automático para administrar de manera eficiente la capacidad de cómputo y lograr un menor costo. **importante** KCL puede recopilar las métricas de uso de la CPU de los procesos de trabajo únicamente si se cumplen ciertos requisitos previos. Para obtener más información, consulte [Requisitos previos](develop-kcl-consumers-java.md#develop-kcl-consumers-java-prerequisites). Si KCL no puede recopilar las métricas de uso de la CPU de los procesos de trabajo, volverá a utilizar el rendimiento por proceso de trabajo para asignar los arrendamientos y equilibrar la carga entre los procesos de trabajo de la flota. KCL supervisará el rendimiento que recibe cada proceso de trabajo en un momento dado y reasignará los arrendamientos para asegurarse de que cada uno obtenga un nivel de rendimiento total similar en los arrendamientos asignados. # Desarrollar consumidores con KCL Puede utilizar Kinesis Client Library (KCL) para crear aplicaciones de consumo que procesen datos de los flujos de datos de Kinesis. KCL está disponible en varios lenguajes. En este tema se describe cómo desarrollar consumidores de KCL en lenguajes Java y otros distintos a Java. + Para ver la referencia de Javadoc de la Kinesis Client Library, consulte [Javadoc de Amazon Kinesis Client Library](https://javadoc.io/doc/software.amazon.kinesis/amazon-kinesis-client/latest/index.html). + Para descargar KCL para Java desde GitHub, consulte la [biblioteca de clientes de Amazon Kinesis](https://github.com/awslabs/amazon-kinesis-client) para Java. + Para localizar la KCL para Java en Apache Maven, consulte el [Repositorio central de KCL Maven](https://central.sonatype.com/artifact/software.amazon.kinesis/amazon-kinesis-client). **Topics** + [ # Desarrollar consumidores con KCL en Java ](develop-kcl-consumers-java.md) + [ # Desarrollar consumidores con KCL en lenguajes distintos de Java ](develop-kcl-consumers-non-java.md) # Desarrollar consumidores con KCL en Java ## Requisitos previos Antes de comenzar con KCL 3.x, asegúrese de que dispone de lo siguiente: + Java Development Kit (JDK) 8 o posterior, + AWS SDK para Java 2.x + Maven o Gradle para la administración de dependencias. KCL recopila métricas de uso de la CPU, como el uso de la CPU del host de cómputo en el que se ejecutan los procesos de trabajo para equilibrar la carga y lograr un nivel de uso de recursos uniforme entre ellos. Para permitir que KCL recopile métricas de uso de CPU desde los procesos de trabajo, debe cumplir los siguientes requisitos previos: **Amazon Elastic Compute Cloud(Amazon EC2)** + El sistema operativo debe ser Linux. + Debe habilitarlo [IMDSv2](https://docs.aws.amazon.com/AWSEC2/latest/UserGuide/configuring-instance-metadata-service.html)en su instancia EC2. **Amazon Elastic Container Service (Amazon ECS) en Amazon EC2** + El sistema operativo debe ser Linux. + Debe habilitar la [versión 4 del punto de conexión de metadatos de tareas de ECS](https://docs.aws.amazon.com/AmazonECS/latest/developerguide/ec2-metadata.html). + La versión del agente de contenedor de Amazon ECS debe ser 1.39.0 o posterior. **Amazon ECS en AWS Fargate** + Debe habilitar la [versión 4 del punto de conexión de metadatos de tareas de Fargate](https://docs.aws.amazon.com/AmazonECS/latest/developerguide/task-metadata-endpoint-v4-fargate.html). Si utiliza la versión 1.4.0 o una posterior de la plataforma Fargate, se habilitará de forma predeterminada. + Versión de la plataforma de Fargate 1.4.0 o posterior. **Amazon Elastic Kubernetes Service (Amazon EKS) en Amazon EC2** + El sistema operativo debe ser Linux. **Amazon EKS en AWS Fargate** + Plataforma de Fargate 1.3.0 o posterior. **importante** Si KCL no puede recopilar las métricas de uso de la CPU de los procesos de trabajo, volverá a utilizar el rendimiento por proceso de trabajo para asignar los arrendamientos y equilibrar la carga entre los procesos de trabajo de la flota. Para obtener más información, consulte [Cómo KCL asigna los arrendamientos a los procesos de trabajo y equilibra la carga](kcl-dynamoDB.md#kcl-assign-leases). ## Instale y agrege dependencias Si está utilizando Maven, agregue la siguiente dependencia a su archivo `pom.xml`. Asegúrese de reemplazar la versión 3.x.x por la versión más reciente de KCL. ``` software.amazon.kinesis amazon-kinesis-client 3.x.x ``` Si está utilizando Gradle, agregue lo siguiente a su archivo `build.gradle`. Asegúrese de reemplazar la versión 3.x.x por la versión más reciente de KCL. ``` implementation 'software.amazon.kinesis:amazon-kinesis-client:3.x.x' ``` Puede buscar la versión más reciente del KCL en el [Repositorio central de Maven](https://search.maven.org/artifact/software.amazon.kinesis/amazon-kinesis-client). ## Implementar el consumidor Una aplicación de consumidor de KCL consta de los siguientes componentes clave: **Topics** + [ ### RecordProcessor ](#implementation-recordprocessor) + [ ### RecordProcessorFactory ](#implementation-recordprocessorfactory) + [ ### Programador ](#implementation-scheduler) + [ ### Aplicación de consumo principal ](#implementation-main) ### RecordProcessor RecordProcessor es el componente principal en el que reside la lógica empresarial para procesar los registros de transmisión de datos de Kinesis. Define la forma en que procesa la aplicación los datos que recibe del flujo de Kinesis. Responsabilidades principales: + Inicializar el procesamiento de una partición + Procesar lotes de registros del flujo de Kinesis + Cerrar el procesamiento de una partición (por ejemplo, cuando la partición se divide o fusiona, o cuando el arrendamiento se transfiere a otro host) + Controlar el registro de puntos de control para realizar un seguimiento del progreso A continuación, se muestra un ejemplo de implementación: ``` import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.slf4j.MDC; import software.amazon.kinesis.exceptions.InvalidStateException; import software.amazon.kinesis.exceptions.ShutdownException; import software.amazon.kinesis.lifecycle.events.*; import software.amazon.kinesis.processor.ShardRecordProcessor; public class SampleRecordProcessor implements ShardRecordProcessor { private static final String SHARD_ID_MDC_KEY = "ShardId"; private static final Logger log = LoggerFactory.getLogger(SampleRecordProcessor.class); private String shardId; @Override public void initialize(InitializationInput initializationInput) { shardId = initializationInput.shardId(); MDC.put(SHARD_ID_MDC_KEY, shardId); try { log.info("Initializing @ Sequence: {}", initializationInput.extendedSequenceNumber()); } finally { MDC.remove(SHARD_ID_MDC_KEY); } } @Override public void processRecords(ProcessRecordsInput processRecordsInput) { MDC.put(SHARD_ID_MDC_KEY, shardId); try { log.info("Processing {} record(s)", processRecordsInput.records().size()); processRecordsInput.records().forEach(r -> log.info("Processing record pk: {} -- Seq: {}", r.partitionKey(), r.sequenceNumber()) ); // Checkpoint periodically processRecordsInput.checkpointer().checkpoint(); } catch (Throwable t) { log.error("Caught throwable while processing records. Aborting.", t); } finally { MDC.remove(SHARD_ID_MDC_KEY); } } @Override public void leaseLost(LeaseLostInput leaseLostInput) { MDC.put(SHARD_ID_MDC_KEY, shardId); try { log.info("Lost lease, so terminating."); } finally { MDC.remove(SHARD_ID_MDC_KEY); } } @Override public void shardEnded(ShardEndedInput shardEndedInput) { MDC.put(SHARD_ID_MDC_KEY, shardId); try { log.info("Reached shard end checkpointing."); shardEndedInput.checkpointer().checkpoint(); } catch (ShutdownException | InvalidStateException e) { log.error("Exception while checkpointing at shard end. Giving up.", e); } finally { MDC.remove(SHARD_ID_MDC_KEY); } } @Override public void shutdownRequested(ShutdownRequestedInput shutdownRequestedInput) { MDC.put(SHARD_ID_MDC_KEY, shardId); try { log.info("Scheduler is shutting down, checkpointing."); shutdownRequestedInput.checkpointer().checkpoint(); } catch (ShutdownException | InvalidStateException e) { log.error("Exception while checkpointing at requested shutdown. Giving up.", e); } finally { MDC.remove(SHARD_ID_MDC_KEY); } } } ``` Ahora se explicará con detalle cada método utilizado en el ejemplo: **inicializar (InitializationInput) InitializationInput** + Objetivo: configurar los recursos o estados necesarios para procesar los registros. + Cuándo se llama: una vez, cuando KCL asigna una partición a este procesador de registros. + Puntos clave: + `initializationInput.shardId()`: el ID de la partición que gestionará este procesador. + `initializationInput.extendedSequenceNumber()`: el número de secuencia desde el que se iniciará el procesamiento. **ProcessRecords () ProcessRecordsInput processRecordsInput** + Objetivo: procesar los registros entrantes y, de manera opcional, comprobar el progreso del punto de control. + Cuando se llama: repetidamente, siempre y cuando el procesador de registros sea el propietario del arrendamiento de la partición. + Puntos clave: + `processRecordsInput.records()`: lista de registros que se van a procesar. + `processRecordsInput.checkpointer()`: se utiliza para comprobar el progreso. + Asegúrese de haber gestionado cualquier excepción durante el procesamiento para evitar que la KCL falle. + Este método debe ser idempotente, ya que el mismo registro puede procesarse más de una vez en algunos escenarios, por ejemplo, cuando los datos no se han registrado en puntos de control antes de que el proceso de trabajo fallara o reinicie de forma inesperada. + Vacíe siempre los datos almacenados en el búfer antes de registrar los puntos de control para garantizar la coherencia de datos. **Arrendamiento perdido () LeaseLostInput leaseLostInput** + Objetivo: limpiar cualquier recurso específico para procesar esta partición. + Cuándo se llama: cuando otro programador se hace cargo del arrendamiento de esta partición. + Puntos clave: + El registro de puntos de control no está permitido en este método. **Fragmentado () ShardEndedInput shardEndedInput** + Objetivo: finalizar el procesamiento de esta partición y este punto de control. + Cuándo se llama: cuando la partición se divide o se fusiona, lo que indica que se han procesado todos los datos de esta partición. + Puntos clave: + `shardEndedInput.checkpointer()`: se utiliza para realizar el registro final de puntos de control. + El registro de puntos de control de este método es obligatorio para completar el procesamiento. + Si no se vacían los datos y puntos de control, es posible que se pierdan los datos o se duplique el procesamiento cuando se vuelva a abrir la partición. **Cierre solicitado () ShutdownRequestedInput shutdownRequestedInput** + Objetivo: registrar un punto de control y limpiar los recursos cuando KCL se está cerrando. + Cuándo se llama: cuando KCL se cierra (por ejemplo, cuando la aplicación se cierra). + Puntos clave: + `shutdownRequestedInput.checkpointer()`: se utiliza para realizar el registro de puntos de control antes del cierre. + Asegúrese de haber implementado el registro de puntos de control en el método para guardar el progreso antes de que la aplicación se detenga. + Si no se vacían los datos y puntos de control, se podrían perder los datos o volver a procesar los registros cuando se reinicie la aplicación. **importante** KCL 3.x garantiza un menor reprocesamiento de datos cuando el arrendamiento se transfiere de un proceso de trabajo a otro mediante puntos de control antes de que el proceso de trabajo anterior se cierre. Si no implementa la lógica de registro de puntos de control en el método `shutdownRequested()`, no verá este beneficio. Asegúrese de haber implementado una lógica de registro de puntos de control dentro del método `shutdownRequested()`. ### RecordProcessorFactory RecordProcessorFactory es responsable de crear nuevas RecordProcessor instancias. KCL usa esta fábrica para crear una nueva RecordProcessor para cada fragmento que la aplicación necesite procesar. Responsabilidades principales: + Cree nuevas RecordProcessor instancias bajo demanda + Asegúrese de que cada una RecordProcessor esté inicializada correctamente A continuación, se muestra un ejemplo de implementación: ``` import software.amazon.kinesis.processor.ShardRecordProcessor; import software.amazon.kinesis.processor.ShardRecordProcessorFactory; public class SampleRecordProcessorFactory implements ShardRecordProcessorFactory { @Override public ShardRecordProcessor shardRecordProcessor() { return new SampleRecordProcessor(); } } ``` En este ejemplo, la fábrica crea una nueva SampleRecordProcessor cada vez que se llama a shardRecordProcessor (). Puede ampliarlo para incluir cualquier lógica de inicialización necesaria. ### Programador El programador es un componente de alto nivel que coordina todas las actividades de la aplicación KCL. Es responsable de la orquestación general del procesamiento de datos. Responsabilidades principales: + Gestione el ciclo de vida de RecordProcessors + Gestionar la administración de los arrendamientos de particiones + Coordinar el registro de puntos de control + Equilibrar la carga de procesamiento de la partición entre varios procesos de trabajo de su aplicación + Gestionar correctamente las señales de cierre y cierre de las aplicaciones Por lo general, el programar se crea en la aplicación principal y se inicia en ella. Puede consultar el ejemplo de implementación del programador en la siguiente sección, Aplicación de consumo principal ### Aplicación de consumo principal La aplicación de consumo principal une todos los componentes. Es responsable de configurar el consumo de KCL, crear los clientes necesarios, configurar el programador y administrar el ciclo de vida de la aplicación. Responsabilidades principales: + Configurar clientes de AWS servicio (Kinesis, DynamoDB,) CloudWatch + Configurar la aplicación de KCL + Crear e iniciar el programador + Gestionar el apagado de aplicaciones A continuación, se muestra un ejemplo de implementación: ``` import software.amazon.awssdk.regions.Region; import software.amazon.awssdk.services.cloudwatch.CloudWatchAsyncClient; import software.amazon.awssdk.services.dynamodb.DynamoDbAsyncClient; import software.amazon.awssdk.services.kinesis.KinesisAsyncClient; import software.amazon.kinesis.common.ConfigsBuilder; import software.amazon.kinesis.common.KinesisClientUtil; import software.amazon.kinesis.coordinator.Scheduler; import java.util.UUID; public class SampleConsumer { private final String streamName; private final Region region; private final KinesisAsyncClient kinesisClient; public SampleConsumer(String streamName, Region region) { this.streamName = streamName; this.region = region; this.kinesisClient = KinesisClientUtil.createKinesisAsyncClient(KinesisAsyncClient.builder().region(this.region)); } public void run() { DynamoDbAsyncClient dynamoDbAsyncClient = DynamoDbAsyncClient.builder().region(region).build(); CloudWatchAsyncClient cloudWatchClient = CloudWatchAsyncClient.builder().region(region).build(); ConfigsBuilder configsBuilder = new ConfigsBuilder( streamName, streamName, kinesisClient, dynamoDbAsyncClient, cloudWatchClient, UUID.randomUUID().toString(), new SampleRecordProcessorFactory() ); Scheduler scheduler = new Scheduler( configsBuilder.checkpointConfig(), configsBuilder.coordinatorConfig(), configsBuilder.leaseManagementConfig(), configsBuilder.lifecycleConfig(), configsBuilder.metricsConfig(), configsBuilder.processorConfig(), configsBuilder.retrievalConfig() ); Thread schedulerThread = new Thread(scheduler); schedulerThread.setDaemon(true); schedulerThread.start(); } public static void main(String[] args) { String streamName = "your-stream-name"; // replace with your stream name Region region = Region.US_EAST_1; // replace with your region new SampleConsumer(streamName, region).run(); } } ``` Por defecto, KCL crea un consumidor de distribución ramificada mejorada (EFO) con un rendimiento dedicado. Para obtener más información sobre la distribución ramificada mejorada, consulte [Desarrollo de consumidores de distribución ramificada mejorada con rendimiento dedicado](enhanced-consumers.md). Si tiene menos de 2 consumidores o no necesita retrasos de propagación de la lectura inferiores a 200 ms, debe establecer la siguiente configuración en el objeto del programador para utilizar consumidores de rendimiento compartido: ``` configsBuilder.retrievalConfig().retrievalSpecificConfig(new PollingConfig(streamName, kinesisClient)) ``` El siguiente código es un ejemplo de cómo crear un objeto del programador que utiliza consumidores de rendimiento compartido: **Importaciones:** ``` import software.amazon.kinesis.retrieval.polling.PollingConfig; ``` **Código**: ``` Scheduler scheduler = new Scheduler( configsBuilder.checkpointConfig(), configsBuilder.coordinatorConfig(), configsBuilder.leaseManagementConfig(), configsBuilder.lifecycleConfig(), configsBuilder.metricsConfig(), configsBuilder.processorConfig(), configsBuilder.retrievalConfig().retrievalSpecificConfig(new PollingConfig(streamName, kinesisClient)) );/ ``` # Desarrollar consumidores con KCL en lenguajes distintos de Java En esta sección se describe la implementación de los consumidores que utilizan Kinesis Client Library (KCL) en Python, Node.js, .NET y Ruby. KCL es una biblioteca de Java. El soporte para lenguajes distintos de Java se proporciona mediante una interfaz multilingüe llamada `MultiLangDaemon`. Este daemon está basado en Java y se ejecuta en segundo plano cuando se utiliza con un lenguaje de KCL distinto de Java. Por tanto, si instala KCL para lenguajes distintos de Java y escribe completamente su aplicación de consumo en lenguajes distintos de Java, seguirá necesitando tener Java instalado en su sistema debido al `MultiLangDaemon`. Además, `MultiLangDaemon` tiene algunos ajustes predeterminados que podría tener que personalizar para su caso de uso (por ejemplo, la región de AWS a la que se conecta). Para obtener más información `MultiLangDaemon` sobre él GitHub, consulte el [ MultiLangDaemon proyecto KCL](https://github.com/awslabs/amazon-kinesis-client/tree/v1.x/src/main/java/com/amazonaws/services/kinesis/multilang). Si bien los conceptos principales siguen siendo los mismos en todos los lenguajes, existen algunas consideraciones e implementaciones específicas de cada uno. Para conocer los conceptos básicos sobre el desarrollo de los consumidores de KCL, consulte [Desarrollar consumidores con KCL en Java](develop-kcl-consumers-java.md). Para obtener información más detallada sobre cómo desarrollar consumidores de KCL en Python, Node.js, .NET y Ruby y las últimas actualizaciones, consulte los siguientes GitHub repositorios: + Python: [amazon-kinesis-client-python](https://github.com/awslabs/amazon-kinesis-client-python) + Node.js: [amazon-kinesis-client-nodejs](https://github.com/awslabs/amazon-kinesis-client-nodejs) + .NET: [amazon-kinesis-client-net](https://github.com/awslabs/amazon-kinesis-client-net) + Ruby: [amazon-kinesis-client-ruby](https://github.com/awslabs/amazon-kinesis-client-ruby) **importante** No utilice las siguientes versiones de la biblioteca KCL que no sean de Java si utiliza JDK 8. Estas versiones contienen una dependencia (logback) que es incompatible con JDK 8. KCL Python 3.0.2 y 2.2.0 KCL Node.js 2.3.0 KCL .NET 3.1.0 KCL Ruby 2.2.0 Recomendamos utilizar versiones publicadas antes o después de estas versiones afectadas cuando trabaje con JDK 8. # Procesamiento de varios flujos con KCL En esta sección se describen los cambios necesarios en KCL, que permiten crear aplicaciones de consumo de KCL que pueden procesar más de un flujo de datos al mismo tiempo. **importante** El procesamiento de varios flujos solo es compatible con KCL 2.3 o versiones posteriores. Los consumidores de KCL que estén escritos en lenguajes distintos de Java y que funcionen con `multilangdaemon` *no* son compatibles con el procesamiento de varios flujos. El procesamiento de varios flujos *no* es compatible con ninguna versión de KCL 1.x. + **MultistreamTracker interfaz** + Para crear una aplicación de consumo que pueda procesar múltiples transmisiones al mismo tiempo, debe implementar una nueva interfaz llamada [MultistreamTracker](https://github.com/awslabs/amazon-kinesis-client/blob/0c5042dadf794fe988438436252a5a8fe70b6b0b/amazon-kinesis-client/src/main/java/software/amazon/kinesis/processor/MultiStreamTracker.java). Esta interfaz incluye el método `streamConfigList` que devuelve la lista de flujos de datos y sus configuraciones para que los procese la aplicación de consumo de KCL. Tenga en cuenta que los flujos de datos que se procesan pueden cambiar durante el tiempo de ejecución de la aplicación de consumo. KCL llama a `streamConfigList` periódicamente para obtener información sobre los cambios en los flujos de datos que se van a procesar. + `streamConfigList`Rellena la [StreamConfig](https://github.com/awslabs/amazon-kinesis-client/blob/0c5042dadf794fe988438436252a5a8fe70b6b0b/amazon-kinesis-client/src/main/java/software/amazon/kinesis/common/StreamConfig.java#L23)lista. ``` package software.amazon.kinesis.common; import lombok.Data; import lombok.experimental.Accessors; @Data @Accessors(fluent = true) public class StreamConfig { private final StreamIdentifier streamIdentifier; private final InitialPositionInStreamExtended initialPositionInStreamExtended; private String consumerArn; } ``` + Los campos `StreamIdentifier` y `InitialPositionInStreamExtended` son obligatorios, aunque `consumerArn` es opcional. Debe proporcionar `consumerArn` únicamente si utiliza KCL para implementar una aplicación de consumo con distribución ramificada mejorada. + Para obtener más información al respecto`StreamIdentifier`, consulte [https://github.com/awslabs/amazon-kinesis-client/blob/v2.5.8/amazon-kinesis-client/src/main/java/software/amazon/kinesis/common/StreamIdentifier.java \$1L129.](https://github.com/awslabs/amazon-kinesis-client/blob/v2.5.8/amazon-kinesis-client/src/main/java/software/amazon/kinesis/common/StreamIdentifier.java#L129) Para crear un `StreamIdentifier`, le recomendamos que cree una instancia multiflujo a partir de `streamArn` y la `streamCreationEpoch` que esté disponible en KCL 2.5.0 o versiones posteriores. En las versiones KCL 2.3 y 2.4, que no son compatibles con `streamArm`, cree una instancia multiflujo con el formato `account-id:StreamName:streamCreationTimestamp`. Este formato quedará obsoleto y dejará de ser compatible a partir de la próxima versión principal. + MultistreamTracker también incluye una estrategia para eliminar las concesiones de transmisiones antiguas de la tabla de arrendamientos () formerStreamsLeasesDeletionStrategy. Tenga en cuenta que la estrategia NO SE PUEDE cambiar durante el tiempo de ejecución de la aplicación de consumo. Para obtener más información, consulte [https://github.com/awslabs/amazon-kinesis-clientb/amazon-kinesis-client/src/main/java/software/amazon/kinesis/processor/FormerStreamsLeasesDeletionStrategy/blob/0c5042dadf794fe988438436252a5a8fe70b6b0](https://github.com/awslabs/amazon-kinesis-client/blob/0c5042dadf794fe988438436252a5a8fe70b6b0b/amazon-kinesis-client/src/main/java/software/amazon/kinesis/processor/FormerStreamsLeasesDeletionStrategy.java) .java. + [ConfigsBuilder](https://github.com/awslabs/amazon-kinesis-client/blob/0c5042dadf794fe988438436252a5a8fe70b6b0b/amazon-kinesis-client/src/main/java/software/amazon/kinesis/common/ConfigsBuilder.java)es una clase que abarca toda la aplicación y que se puede utilizar para especificar todos los ajustes de configuración de KCL que se utilizarán al crear la aplicación de consumo de KCL para la versión 2.x o posterior de KCL. `ConfigsBuilder`la clase ahora es compatible con la interfaz. `MultistreamTracker` Puede inicializar ConfigsBuilder cualquiera de las dos con el nombre del flujo de datos desde el que se van a consumir los registros: ``` /** * Constructor to initialize ConfigsBuilder with StreamName * @param streamName * @param applicationName * @param kinesisClient * @param dynamoDBClient * @param cloudWatchClient * @param workerIdentifier * @param shardRecordProcessorFactory */ public ConfigsBuilder(@NonNull String streamName, @NonNull String applicationName, @NonNull KinesisAsyncClient kinesisClient, @NonNull DynamoDbAsyncClient dynamoDBClient, @NonNull CloudWatchAsyncClient cloudWatchClient, @NonNull String workerIdentifier, @NonNull ShardRecordProcessorFactory shardRecordProcessorFactory) { this.appStreamTracker = Either.right(streamName); this.applicationName = applicationName; this.kinesisClient = kinesisClient; this.dynamoDBClient = dynamoDBClient; this.cloudWatchClient = cloudWatchClient; this.workerIdentifier = workerIdentifier; this.shardRecordProcessorFactory = shardRecordProcessorFactory; } ``` O bien, puede inicializarlo ConfigsBuilder `MultiStreamTracker` si desea implementar una aplicación de consumo de KCL que procese varios flujos al mismo tiempo. ``` * Constructor to initialize ConfigsBuilder with MultiStreamTracker * @param multiStreamTracker * @param applicationName * @param kinesisClient * @param dynamoDBClient * @param cloudWatchClient * @param workerIdentifier * @param shardRecordProcessorFactory */ public ConfigsBuilder(@NonNull MultiStreamTracker multiStreamTracker, @NonNull String applicationName, @NonNull KinesisAsyncClient kinesisClient, @NonNull DynamoDbAsyncClient dynamoDBClient, @NonNull CloudWatchAsyncClient cloudWatchClient, @NonNull String workerIdentifier, @NonNull ShardRecordProcessorFactory shardRecordProcessorFactory) { this.appStreamTracker = Either.left(multiStreamTracker); this.applicationName = applicationName; this.kinesisClient = kinesisClient; this.dynamoDBClient = dynamoDBClient; this.cloudWatchClient = cloudWatchClient; this.workerIdentifier = workerIdentifier; this.shardRecordProcessorFactory = shardRecordProcessorFactory; } ``` + Con la compatibilidad con varios flujos implementada para la aplicación de consumo de KCL, cada fila de la tabla de arrendamiento de la aplicación ahora contiene el ID de la partición y el nombre del flujo de los varios flujos de datos que procesa esta aplicación. + Cuando se implementa la compatibilidad con varios flujos para la aplicación de consumo de KCL, leaseKey adopta la siguiente estructura: `account-id:StreamName:streamCreationTimestamp:ShardId`. Por ejemplo, `111111111:multiStreamTest-1:12345:shardId-000000000336`. **importante** Cuando la aplicación de consumo de KCL existente está configurada para procesar solo un flujo de datos, `leaseKey` (que es la clave de partición de la tabla de arrendamiento) es el ID de la partición. Si vuelve a configurar una aplicación de consumo de KCL existente para procesar varios flujos de datos, se rompe la tabla de arrendamiento, ya que la estructura de `leaseKey` debe ser la siguiente: `account-id:StreamName:StreamCreationTimestamp:ShardId` para admitir varios flujos. # Utilice el registro de AWS Glue esquemas con KCL Puede integrar Kinesis Data Streams con AWS Glue el registro de esquemas. El registro de AWS Glue esquemas le permite descubrir, controlar y desarrollar esquemas de forma centralizada, al tiempo que garantiza que los datos generados se validen continuamente mediante un esquema registrado. 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 [Registro de esquemas de AWS Glue](https://docs.aws.amazon.com/glue/latest/dg/schema-registry.html). Una de las formas de configurar esta integración es a través de KCL para Java. **importante** AWS Glue La integración del registro de esquemas para Kinesis Data Streams solo se admite en KCL 2.3 o versiones posteriores. AWS Glue La integración del registro de esquemas para Kinesis Data Streams no *es* compatible con los consumidores de KCL escritos en lenguajes distintos de Java que se ejecutan con. `multilangdaemon` AWS Glue La integración del registro de esquemas para Kinesis Data Streams no *es* compatible con ninguna versión de KCL 1.x. Para obtener instrucciones detalladas sobre cómo configurar la integración de Kinesis Data Streams con el registro de AWS Glue esquemas mediante KCL, consulte la sección «Interacción con los datos mediante las KPL/KCL bibliotecas» en [Caso de uso: integración de Amazon Kinesis Data Streams](https://docs.aws.amazon.com/glue/latest/dg/schema-registry-integrations.html#schema-registry-integrations-kds) con el registro de esquemas. AWS Glue # Permisos de IAM necesarios para las aplicaciones de consumo de KCL Debe agregar los siguientes permisos al rol o usuario de IAM asociado a su aplicación de consumo de KCL. Prácticas recomendadas de seguridad para AWS dictar el uso de permisos detallados para controlar el acceso a los distintos recursos. AWS Identity and Access Management (IAM) le permite administrar los usuarios y los permisos de acceso de los usuarios. AWS Una política de IAM enumera de forma explícita las acciones que se pueden realizar y los recursos a los que se pueden aplicar dichas acciones. En la siguiente tabla se muestran los permisos de IAM mínimos que generalmente se requieren para las aplicaciones de consumo de KCL: **Permisos de IAM mínimos para las aplicaciones de consumo de KCL** | Servicio | Acciones | Recursos () ARNs | Finalidad | | --- | --- | --- | --- | | Amazon Kinesis Data Streams | `DescribeStream` `DescribeStreamSummary` `RegisterStreamConsumer` | El flujo de datos de Kinesis desde la que la aplicación de KCL procesará los datos.`arn:aws:kinesis:region:account:stream/StreamName` | Antes de intentar leer registros, el consumidor comprueba si la secuencia de datos existe y si está activa, y si los fragmentos se encuentran en la secuencia de datos. Registra los consumidores en una partición. | | Amazon Kinesis Data Streams | `GetRecords` `GetShardIterator` `ListShards` | El flujo de datos de Kinesis desde la que la aplicación de KCL procesará los datos.`arn:aws:kinesis:region:account:stream/StreamName` | Lee registros de un fragmento. | | Amazon Kinesis Data Streams | `SubscribeToShard` `DescribeStreamConsumer` | El flujo de datos de Kinesis desde la que la aplicación de KCL procesará los datos. Agregue esta acción solo si utiliza consumidores con distribución ramificada mejorada (EFO). `arn:aws:kinesis:region:account:stream/StreamName/consumer/*` | Se suscribe a una sección destinada a los consumidores con distribución ramificada mejorada (EFO). | | Amazon DynamoDB | `CreateTable` `DescribeTable` `UpdateTable` `Scan` `GetItem` `PutItem` `UpdateItem` `DeleteItem` | Tabla de arrendamiento (tabla de metadatos en DynamoDB creada por KCL). `arn:aws:dynamodb:region:account:table/KCLApplicationName` | Estas acciones son necesarias para que KCL administre la tabla de arrendamiento creada en DynamoDB. | | Amazon DynamoDB | `CreateTable` `DescribeTable` `Scan` `GetItem` `PutItem` `UpdateItem` `DeleteItem` | Tabla de métricas de procesos de trabajo y de estado de coordinadores (tablas de metadatos en DynamoDB) creadas por KCL. `arn:aws:dynamodb:region:account:table/KCLApplicationName-WorkerMetricStats` `arn:aws:dynamodb:region:account:table/KCLApplicationName-CoordinatorState` | Estas acciones son obligatorias para que KCL administre las tablas de metadatos para las métricas de los procesos de trabajo y del estado de coordinadores en DynamoDB. | | Amazon DynamoDB | `Query` | Un índice secundario global en la tabla de arrendamiento. `arn:aws:dynamodb:region:account:table/KCLApplicationName/index/*` | Esta acción es obligatoria para que KCL lea el índice secundario global de la tabla de arrendamiento creada en DynamoDB. | | Amazon CloudWatch | `PutMetricData` | \$1 | Sube métricas CloudWatch que sean útiles para monitorear la aplicación. El asterisco (\$1) se utiliza porque no hay ningún recurso específico CloudWatch en el que se invoque la `PutMetricData` acción. | **nota** Sustituya «región», «cuenta» y «StreamNamenombre» por su propio Cuenta de AWS número Región de AWS, KCLApplication nombre de la ARNs transmisión de datos de Kinesis y nombre de la aplicación de KCL, respectivamente. KCL 3.x crea dos tablas de metadatos más en DynamoDB. Para obtener más información sobre las tablas de metadatos de DynamoDB creadas por KCL, consulte [Tablas de metadatos de DynamoDB y equilibrio de carga en KCL](kcl-dynamoDB.md). Si utiliza configuraciones para personalizar los nombres de las tablas de metadatos creadas por KCL, utilice los nombres de tabla especificados en lugar del nombre de la aplicación de KCL. A continuación, se muestra un ejemplo de documento de política para una aplicación de consumo de KCL. ------ #### [ JSON ] **** ``` { "Version":"2012-10-17", "Statement": [ { "Effect": "Allow", "Action": [ "kinesis:DescribeStream", "kinesis:DescribeStreamSummary", "kinesis:RegisterStreamConsumer", "kinesis:GetRecords", "kinesis:GetShardIterator", "kinesis:ListShards" ], "Resource": "arn:aws:kinesis:us-east-1:123456789012:stream/STREAM_NAME" }, { "Effect": "Allow", "Action": [ "kinesis:SubscribeToShard", "kinesis:DescribeStreamConsumer" ], "Resource": "arn:aws:kinesis:us-east-1:123456789012:stream/STREAM_NAME/consumer/*" }, { "Effect": "Allow", "Action": [ "dynamodb:CreateTable", "dynamodb:DescribeTable", "dynamodb:UpdateTable", "dynamodb:GetItem", "dynamodb:UpdateItem", "dynamodb:PutItem", "dynamodb:DeleteItem", "dynamodb:Scan" ], "Resource": [ "arn:aws:dynamodb:us-east-1:123456789012:table/KCL_APPLICATION_NAME" ] }, { "Effect": "Allow", "Action": [ "dynamodb:CreateTable", "dynamodb:DescribeTable", "dynamodb:GetItem", "dynamodb:UpdateItem", "dynamodb:PutItem", "dynamodb:DeleteItem", "dynamodb:Scan" ], "Resource": [ "arn:aws:dynamodb:us-east-1:123456789012:table/KCL_APPLICATION_NAME-WorkerMetricStats", "arn:aws:dynamodb:us-east-1:123456789012:table/KCL_APPLICATION_NAME-CoordinatorState" ] }, { "Effect": "Allow", "Action": [ "dynamodb:Query" ], "Resource": [ "arn:aws:dynamodb:us-east-1:123456789012:table/KCL_APPLICATION_NAME/index/*" ] }, { "Effect": "Allow", "Action": [ "cloudwatch:PutMetricData" ], "Resource": "*" } ] } ``` ------ Antes de utilizar la política de ejemplo, revise los siguientes elementos: + Sustituya REGION por su Región de AWS (por ejemplo, us-east-1). + Sustituya ACCOUNT\$1ID por su ID. Cuenta de AWS + sustituya STREAM\$1NAME por el nombre de su flujo de datos de Kinesis, + sustituya CONSUMER\$1NAME por el nombre de su consumidor. Suele ser el nombre de su aplicación al utilizar KCL, + sustituya KCL\$1APPLICATION\$1NAME por el nombre de la aplicación de KCL. # Configuraciones de KCL Defina propiedades de configuración para personalizar la funcionalidad de Kinesis Client Library y cumplir requisitos específicos. En la siguiente tabla se describen las propiedades y clases de configuración. **importante** En KCL 3.x, el algoritmo de equilibrio de carga tiene como objetivo lograr un uso uniforme de la CPU entre los procesos de trabajo, y no un número igual de arrendamientos por proceso de trabajo. Si establece `maxLeasesForWorker` demasiado bajo, podría limitar la capacidad de KCL de equilibrar la carga de trabajo de forma eficaz. Si utiliza la configuración `maxLeasesForWorker`, tenga presente aumentar su valor para permitir la mejor distribución de carga posible. **En esta tabla se muestran las propiedades de configuración de KCL** | Propiedad de configuración | Clase de configuración | Description (Descripción) | Predeterminado | | --- | --- | --- | --- | | applicationName | ConfigsBuilder | El nombre de la aplicación de KCL. Se utiliza de forma predeterminada para tableName y consumerName. | No aplicable | | tableName | ConfigsBuilder | Permite sustituir el nombre de la tabla que se utiliza para la tabla de asignaciones de Amazon DynamoDB. | No aplicable | | streamName | ConfigsBuilder | El nombre de la secuencia cuyos registros procesa de esta aplicación. | No aplicable | | workerIdentifier | ConfigsBuilder | Identificador único que representa esta instancia del procesador de aplicaciones. Deben ser único. | No aplicable | | failoverTimeMillis | LeaseManagementConfig | El número de milisegundos que deben transcurrir antes de que se pueda considerar que se ha producido un error en el propietario de una asignación. En el caso de las aplicaciones que tienen una gran cantidad de particiones, se pueden establecer en un número mayor para reducir la cantidad de IOPS de DynamoDB necesarios para el seguimiento de los arrendamientos. | 10 000 (10 segundos) | | shardSyncIntervalMillis | LeaseManagementConfig | El tiempo entre llamadas de sincronización del fragmento. | 60 000 (60 segundos) | | cleanupLeasesUponShardCompletion | LeaseManagementConfig | Cuando se establece, se eliminan las asignaciones tan pronto como se inicia el procesamiento del fragmento secundario. | TRUE | | ignoreUnexpectedChildShards | LeaseManagementConfig | Cuando se establece, los fragmentos secundarios que tienen un fragmento abierto se pasan por alto. Esto es principalmente para DynamoDB Streams. | FALSO | | maxLeasesForWorker | LeaseManagementConfig | La cantidad máxima de arrendamientos que debe aceptar un solo proceso de trabajo. Si se establece un valor demasiado bajo, podría causar que se pierdan datos si los procesos de trabajo no pueden procesar todas las particiones y que la asignación del arrendamiento entre los procesos de trabajo no sea óptima. Tenga en cuenta el número total de particiones, el número de procesos de trabajo y la capacidad de procesamiento de los procesos de trabajo al configurarlo. | Sin límite | | maxLeaseRenewalThreads | LeaseManagementConfig | Controla el tamaño del grupo de subprocesos del renovador de asignaciones. Cuanto más grande sea el número de asignaciones que puede tomar la aplicación, más grande debe ser este grupo. | 20 | | billingMode | LeaseManagementConfig | Determina el modo de capacidad de la tabla de arrendamiento creada en DynamoDB. Existen dos opciones: el modo bajo demanda (PAY\$1PER\$1REQUEST) y el modo aprovisionado. Recomendamos utilizar la configuración predeterminada del modo bajo demanda, ya que escala automáticamente para adaptarse a la carga de trabajo sin necesidad de planificar la capacidad. | PAY\$1PER\$1REQUEST (modo bajo demanda) | | initialLeaseTableReadCapacity | LeaseManagementConfig | Capacidad de lectura de DynamoDB que se utiliza cuando Kinesis Client Library debe crear una tabla de arrendamiento de DynamoDB con el modo de capacidad aprovisionada. Puede ignorar esta configuración si utiliza el modo de capacidad bajo demanda predeterminado en la configuración billingMode. | 10 | | initialLeaseTableWriteCapacity | LeaseManagementConfig | Capacidad de lectura de DynamoDB que se utiliza cuando Kinesis Client Library debe crear una tabla de arrendamiento de DynamoDB. Puede ignorar esta configuración si utiliza el modo de capacidad bajo demanda predeterminado en la configuración billingMode. | 10 | | initialPositionInStreamExtended | LeaseManagementConfig | La posición inicial de la secuencia en la que debería comenzar la aplicación. Esto solo se utiliza durante la creación inicial de la asignación. | InitialPositionInStream.TRIM\$1HORIZON | | reBalanceThresholdPercentage | LeaseManagementConfig | Un valor porcentual que determina cuándo el algoritmo de equilibrio de carga debería reasignar las particiones entre los procesos de trabajo. Es una nueva configuración presentada en KCL 3.x. | 10 | | dampeningPercentage | LeaseManagementConfig | Un valor porcentual que se utiliza para amortiguar la cantidad de carga que se moverá desde el proceso de trabajo sobrecargado en una sola operación de rebalanceo. Es una nueva configuración presentada en KCL 3.x. | 60 | | allowThroughputOvershoot | LeaseManagementConfig | Determina si aún es necesario contratar un arrendamiento adicional al proceso de trabajo sobrecargado, incluso si esto provoca que el monto total del arrendamiento solicitado supere el monto de rendimiento deseado. Es una nueva configuración presentada en KCL 3.x. | TRUE | | disableWorkerMetrics | LeaseManagementConfig | Determina si KCL debe ignorar las métricas de recursos de los procesos de trabajos (como el uso de la CPU) al reasignar los arrendamientos y equilibrar la carga. Configúrelo en TRUE si quiere evitar que KCL equilibre la carga según el uso de la CPU. Es una nueva configuración presentada en KCL 3.x. | FALSO | | maxThroughputPerHostKBps | LeaseManagementConfig | Cantidad de rendimiento máximo que se debe asignar a un proceso de trabajo durante la asignación del arrendamiento. Es una nueva configuración presentada en KCL 3.x. | Sin límite | | isGracefulLeaseHandoffEnabled | LeaseManagementConfig | Controla el comportamiento de la transferencia del arrendamiento entre los procesos de trabajo. Si se establece en true, KCL intentará transferir los arrendamientos sin problemas, dejando que el fragmento RecordProcessor tenga tiempo suficiente para completar el procesamiento antes de entregar el arrendamiento a otro trabajador. Esto puede ayudar a garantizar la integridad de los datos y realizar transiciones fluidas, pero puede aumentar el tiempo de transferencia. Si se establece en falso, el contrato de arrendamiento se transferirá inmediatamente sin esperar a que se cierre correctamente. RecordProcessor Esto puede llevar a que las transferencias sean más rápidas, pero se corre el riesgo de que el procesamiento quede incompleto. Nota: Los puntos de control deben implementarse dentro del método shutdownRequested () del RecordProcessor para poder beneficiarse de la elegante función de traspaso de arrendamientos. Es una nueva configuración presentada en KCL 3.x. | TRUE | | gracefulLeaseHandoffTimeoutMillis | LeaseManagementConfig | Especifica el tiempo mínimo (en milisegundos) que se debe esperar a que el fragmento actual se cierre correctamente antes de RecordProcessor transferir forzosamente el arrendamiento al siguiente propietario. Si el método processRecords suele ejecutarse durante más tiempo que el valor predeterminado, tenga en mente aumentar esta configuración. Esto garantiza que RecordProcessor tenga tiempo suficiente para completar su procesamiento antes de que se produzca la transferencia del arrendamiento. Es una nueva configuración presentada en KCL 3.x. | 30 000 (30 segundos) | | maxRecords | PollingConfig | Permite establecer el número máximo de registros que devuelve Kinesis. | 10 000 | | retryGetRecordsInSeconds | PollingConfig | Configura el retraso entre los GetRecords intentos en caso de error. | Ninguno | | maxGetRecordsThreadPool | PollingConfig | El tamaño del grupo de subprocesos utilizado para GetRecords. | Ninguno | | idleTimeBetweenReadsInMillis | PollingConfig | Determina cuánto tiempo espera KCL entre GetRecords llamadas para sondear los datos de los flujos de datos. La unidad es de milisegundos. | 1500 | | callProcessRecordsEvenForEmptyRecordList | ProcessorConfig | Cuando se establece, se llama al procesador de registros incluso cuando no se proporciona ningún registro de Kinesis. | FALSO | | parentShardPollIntervalMillis | CoordinatorConfig | Determina la frecuencia con que debería sondear un procesador de registros para ver si el fragmento principal se ha completado. La unidad es milisegundos. | 10 000 (10 segundos) | | skipShardSyncAtWorkerInitializationIfLeaseExist | CoordinatorConfig | Deshabilita la sincronización de los datos de los fragmentos si la tabla de asignaciones todavía contiene entradas. | FALSO | | shardPrioritization | CoordinatorConfig | La priorización de fragmentos que se va a utilizar. | NoOpShardPrioritization | | ClientVersionConfig | CoordinatorConfig | Determina en qué modo de compatibilidad de versiones de KCL se ejecutará la aplicación. Esta configuración es solo para la migración desde versiones anteriores de KCL. Al migrar a la versión 3.x, debe establecer esta configuración en `CLIENT_VERSION_CONFIG_COMPATIBLE_WITH_2X`. Puede eliminar esta configuración cuando complete la migración. | CLIENT\$1VERSION\$1CONFIG\$13X | | taskBackoffTimeMillis | LifecycleConfig | El tiempo que se debe esperar para reintentar la tareas de KCL con errores. La unidad es de milisegundos. | 500 (0,5 segundos) | | logWarningForTaskAfterMillis | LifecycleConfig | Tiempo que se debe esperar antes de registrar una advertencia si una tarea no ha finalizado. | Ninguno | | listShardsBackoffTimeInMillis | RetrievalConfig | El número de milisegundos que se debe esperar entre llamadas a ListShards cuando se producen errores. La unidad es de milisegundos. | 1500 (1,5 segundos) | | maxListShardsRetryAttempts | RetrievalConfig | El número máximo de veces que se reintenta ListShards antes de desistir. | 50 | | metricsBufferTimeMillis | MetricsConfig | Especifica el tiempo máximo (en milisegundos) para almacenar en búfer las métricas antes de publicarlas. CloudWatch | 10 000 (10 segundos) | | metricsMaxQueueSize | MetricsConfig | Especifica el número máximo de métricas que se deben almacenar en búfer antes de publicarlas CloudWatch. | 10 000 | | metricsLevel | MetricsConfig | Especifica el nivel de granularidad de CloudWatch las métricas que se van a habilitar y publicar. Valores posibles: NONE, SUMMARY, DETAILED. | MetricsLevel.DETALLADO | | metricsEnabledDimensions | MetricsConfig | Controla las dimensiones permitidas para las CloudWatch métricas. | Todas las dimensiones | **Configuraciones discontinuadas en KCL 3.x** Las siguientes propiedades de configuración están discontinuadas en KCL 3.x: **En la tabla se muestran las propiedades de configuración discontinuadas de KCL 3.x** | Propiedad de configuración | Clase de configuración | Description (Descripción) | | --- | --- | --- | | maxLeasesToStealAtOneTime | LeaseManagementConfig | El número máximo de asignaciones del que debería intentar apropiarse una aplicación al mismo tiempo. KCL 3.x ignorará esta configuración y reasignará los arrendamientos según la utilización de los recursos por parte de los procesos de trabajo. | | enablePriorityLeaseAssignment | LeaseManagementConfig | Controla si los trabajadores deben dar prioridad a los arrendamientos muy vencidos (los arrendamientos no se renuevan hasta triplicar el tiempo de conmutación por error) y a los nuevos arrendamientos de particiones, independiente del número de arrendamientos de destino, pero respetando los límites máximos de arrendamiento. KCL 3.x ignorará esta configuración y siempre distribuirá los arrendamientos vencidos entre los procesos de trabajo. | **importante** Debe seguir teniendo las propiedades de configuración interrumpidas durante la migración de versiones anteriores de KCL a KCL 3.x. Durante la migración, el proceso de trabajo de KCL comenzará primero con el modo compatible con KCL 2.x y pasará al modo de funcionalidad de KCL 3.x cuando detecte que todos los procesos de trabajo de KCL de la aplicación están preparados para ejecutar KCL 3.x. Estas configuraciones discontinuadas son necesarias mientras los procesos de trabajo de KCL utilizan el modo compatible con KCL 2.x. # Política del ciclo de vida de la versión de KCL En este tema se describe la política de ciclo de vida de las versiones para la biblioteca de clientes de Amazon Kinesis (KCL). AWS publica periódicamente nuevas versiones de KCL para incorporar nuevas funciones y mejoras, correcciones de errores, parches de seguridad y actualizaciones de dependencias. Le recomendamos que utilice las versiones de KCL 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 KCL no compatible. El ciclo de vida de las principales versiones de KCL 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 | | --- | --- | --- | --- | --- | | KCL 1.x | Modo de mantenimiento | 19/12/2013 | 2025-04-17 | 2026-01-30 | | KCL 2.x | Disponibilidad general | 2018-08-02 | -- | -- | | KCL 3.x | Disponibilidad general | 2024-11-06 | -- | -- | # Migre desde versiones anteriores de KCL En este tema se explica cómo migrar desde versiones anteriores de Kinesis Client Library (KCL). ## ¿Qué novedades incluye KCL 3.0? Kinesis Client Library (KCL) 3.0 presenta numerosas mejoras importantes en comparación con las versiones anteriores: + Reduce los costos de procesamiento de las aplicaciones de consumo al redistribuir automáticamente el trabajo de los procesos de trabajo sobreutilizados a los procesos de trabajo infrautilizados de la flota de aplicaciones de consumo. Este nuevo algoritmo de equilibrio de carga garantiza una distribución uniforme del uso de la CPU entre los procesos de trabajo y elimina la necesidad de aprovisionar en exceso a los procesos de trabajo. + Reduce el costo de DynamoDB asociado a KCL al optimizar las operaciones de lectura en la tabla de arrendamiento. + Minimiza el reprocesamiento de los datos cuando los arrendamientos se reasignan a otro proceso de trabajo, ya que permite que el proceso de trabajo actual complete el registro de puntos de control de los registros que ha procesado. + Se utiliza AWS SDK for Java 2.x para mejorar el rendimiento y las funciones de seguridad, lo que elimina por completo la dependencia de la versión 1.x. AWS SDK para Java Para obtener más información, consulte las [notas de KCL 3.0](https://github.com/awslabs/amazon-kinesis-client/blob/master/CHANGELOG.md). **Topics** + [ ## ¿Qué novedades incluye KCL 3.0? ](#kcl-migration-new-3-0) + [ # Migre de KCL 2.x a KCL 3.x ](kcl-migration-from-2-3.md) + [ # Restauración de la versión de KCL anterior ](kcl-migration-rollback.md) + [ # Avance a KCL 3.x después de una restauración ](kcl-migration-rollforward.md) + [ # Prácticas recomendadas para la tabla de arrendamiento con el modo de capacidad aprovisionada ](kcl-migration-lease-table.md) + [ # Migración de KCL 1.x a KCL 3.x ](kcl-migration-1-3.md) # Migre de KCL 2.x a KCL 3.x En este tema se proporcionan step-by-step instrucciones para migrar a su consumidor de KCL 2.x a KCL 3.x. KCL 3.x permite la migración local de los consumidores de KCL 2.x. Puede seguir consumiendo los datos de su flujo de datos de Kinesis y, al mismo tiempo, migrar a sus procesos de trabajo de forma continua. **importante** KCL 3.x mantiene las mismas interfaces y métodos que KCL 2.x. Por lo tanto, no es necesario actualizar el código de procesamiento de registros durante la migración. Sin embargo, debe establecer la configuración adecuada y comprobar los pasos necesarios para la migración. Recomendamos encarecidamente que siga los siguientes pasos de migración para que la experiencia de migración sea fluida. ## Paso 1: requisitos previos Antes de comenzar con KCL 3.x, asegúrese de que dispone de lo siguiente: + Java Development Kit (JDK) 8 o posterior, + AWS SDK para Java 2.x + Maven o Gradle para la administración de dependencias. **importante** No utilice las AWS SDK para Java versiones 2.27.19 a 2.27.23 con KCL 3.x. Estas versiones cuentan con un problema que provoca un error de excepción relacionado con el uso de DynamoDB por parte de KCL. Le recomendamos que utilice la AWS SDK para Java versión 2.28.0 o posterior para evitar este problema. ## Paso 2: agregar dependencias Si está utilizando Maven, agregue la siguiente dependencia a su archivo `pom.xml`. Asegúrese de reemplazar la versión 3.x.x por la versión más reciente de KCL. ``` software.amazon.kinesis amazon-kinesis-client 3.x.x ``` Si está utilizando Gradle, agregue lo siguiente a su archivo `build.gradle`. Asegúrese de reemplazar la versión 3.x.x por la versión más reciente de KCL. ``` implementation 'software.amazon.kinesis:amazon-kinesis-client:3.x.x' ``` Puede buscar la versión más reciente del KCL en el [Repositorio central de Maven](https://search.maven.org/artifact/software.amazon.kinesis/amazon-kinesis-client). ## Paso 3: configurar la configuración relacionada con la migración Para migrar de KCL 2.x a KCL 3.x, debe establecer el siguiente parámetro de configuración: + CoordinatorConfig. clientVersionConfig: Esta configuración determina en qué modo de compatibilidad de versiones de KCL se ejecutará la aplicación. Al migrar de KCL 2.x a 3.x, debe establecer esta configuración en `CLIENT_VERSION_CONFIG_COMPATIBLE_WITH_2X`. Para establecer esta configuración, agregue la siguiente línea al crear el objeto del programador: ``` configsBuilder.coordiantorConfig().clientVersionConfig(ClientVersionConfig.CLIENT_VERSION_CONFIG_COMPLATIBLE_WITH_2X) ``` El siguiente es un ejemplo de cómo configurar `CoordinatorConfig.clientVersionConfig` para la migración de KCL 2.x a 3.x. Puede ajustar otras configuraciones según sea necesario según los requisitos específicos: ``` Scheduler scheduler = new Scheduler( configsBuilder.checkpointConfig(), configsBuilder.coordiantorConfig().clientVersionConfig(ClientVersionConfig.CLIENT_VERSION_CONFIG_COMPLATIBLE_WITH_2X), configsBuilder.leaseManagementConfig(), configsBuilder.lifecycleConfig(), configsBuilder.metricsConfig(), configsBuilder.processorConfig(), configsBuilder.retrievalConfig() ); ``` Es importante que todos los procesos de trabajo de la aplicación de consumo utilicen el mismo algoritmo de equilibrio de carga en un momento dado, ya que KCL 2.x y 3.x utilizan diferentes algoritmos de equilibrio de carga. Si los procesos de trabajo utilizan diferentes algoritmos de equilibrio de carga, es posible que la distribución de la carga no sea óptima, ya que los dos algoritmos funcionan de forma independiente. Esta configuración de compatibilidad con KCL 2.x permite que la aplicación KCL 3.x se ejecute en un modo compatible con KCL 2.x y utilice el algoritmo de equilibrio de carga para KCL 2.x hasta que todos los procesos de trabajo de la aplicación de consumo se hayan actualizado a KCL 3.x. Cuando se complete la migración, KCL cambiará automáticamente al modo de funcionalidad completa de KCL 3.x y empezará a utilizar un nuevo algoritmo de equilibrio de carga de KCL 3.x para todos los procesos de trabajo en ejecución. **importante** Si no va a utilizar `ConfigsBuilder`, y va a crear un objeto `LeaseManagementConfig` para establecer las configuraciones, debe agregar otro parámetro denominado `applicationName` en la versión 3.x o posterior de KCL. Para obtener más información, consulte [Error de compilación con el LeaseManagementConfig constructor](https://docs.aws.amazon.com/streams/latest/dev/troubleshooting-consumers.html#compiliation-error-leasemanagementconfig). Recomendamos utilizar `ConfigsBuilder` para establecer las configuraciones de KCL. `ConfigsBuilder` ofrece una forma más flexible y fácil de mantener para configurar su aplicación KCL. ## Paso 4: seguir las prácticas recomendadas para la implementación del método shutdownRequested() KCL 3.x presenta una característica denominada *transferencia ágil de arrendamientos* para minimizar el reprocesamiento de datos cuando se entrega un arrendamiento a otro proceso de trabajo como parte del proceso de reasignación del arrendamiento. Ello se logra mediante el registro de puntos de control en el último número de secuencia procesado en la tabla de arrendamiento antes de la transferencia del arrendamiento. Para garantizar que el traspaso del arrendamiento se realiza correcto, debe asegurarse de invocar el objeto `checkpointer` según el método `shutdownRequested` en su clase `RecordProcessor`. Si no está invocando el objeto `checkpointer` dentro del método `shutdownRequested`, puede implementarlo como se ilustra en el siguiente ejemplo. **importante** El siguiente ejemplo de implementación es un requisito mínimo para que la transferencia del arrendamiento se realice sin contratiempos. Si es necesario, puede ampliarlo para incluir lógica adicional relacionada con los registros de puntos de control. Si va a realizar un procesamiento asíncrono, asegúrese de que todos los registros enviados a la cadena descendente se hayan procesado antes de utilizar el registro de puntos de control. Si bien un traspaso eficiente del arrendamiento reduce considerablemente la probabilidad de que se reprocesen los datos durante las transferencias de arrendamiento, no elimina por completo esta posibilidad. Para preservar la integridad y la coherencia de los datos, diseñe sus aplicaciones de consumo intermedias de manera que sean idempotentes. Esto significa que deberían poder gestionar el posible procesamiento de registros duplicados sin efectos adversos en el sistema en general. ``` /** * Invoked when either Scheduler has been requested to gracefully shutdown * or lease ownership is being transferred gracefully so the current owner * gets one last chance to checkpoint. * * Checkpoints and logs the data a final time. * * @param shutdownRequestedInput Provides access to a checkpointer, allowing a record processor to checkpoint * before the shutdown is completed. */ public void shutdownRequested(ShutdownRequestedInput shutdownRequestedInput) { try { // Ensure that all delivered records are processed // and has been successfully flushed to the downstream before calling // checkpoint // If you are performing any asynchronous processing or flushing to // downstream, you must wait for its completion before invoking // the below checkpoint method. log.info("Scheduler is shutting down, checkpointing."); shutdownRequestedInput.checkpointer().checkpoint(); } catch (ShutdownException | InvalidStateException e) { log.error("Exception while checkpointing at requested shutdown. Giving up.", e); } } ``` ## Paso 5: revisar los requisitos previos de la KCL 3.x para recopilar las métricas de los procesos de trabajo. KCL 3.x recopila métricas de uso de la CPU, como el uso de la CPU por parte de los procesos de trabajo, para equilibrar la carga entre los procesos de trabajo de manera uniforme. Los procesos de trabajo de aplicaciones de consumo se pueden ejecutar en Amazon EC2, Amazon ECS, Amazon EKS o AWS Fargate. KCL 3.x puede recopilar métricas de uso de la CPU de los procesos de trabajo únicamente cuando se cumplan los siguientes requisitos previos: **Amazon Elastic Compute Cloud(Amazon EC2)** + El sistema operativo debe ser Linux. + Debe habilitarlo [IMDSv2](https://docs.aws.amazon.com/AWSEC2/latest/UserGuide/configuring-instance-metadata-service.html)en su instancia de EC2. **Amazon Elastic Container Service (Amazon ECS) en Amazon EC2** + El sistema operativo debe ser Linux. + Debe habilitar la [versión 4 del punto de conexión de metadatos de tareas de ECS](https://docs.aws.amazon.com/AmazonECS/latest/developerguide/ec2-metadata.html). + La versión del agente de contenedor de Amazon ECS debe ser 1.39.0 o posterior. **Amazon ECS en AWS Fargate** + Debe habilitar la [versión 4 del punto de conexión de metadatos de tareas de Fargate](https://docs.aws.amazon.com/AmazonECS/latest/developerguide/task-metadata-endpoint-v4-fargate.html). Si utiliza la versión 1.4.0 o una posterior de la plataforma Fargate, se habilitará de forma predeterminada. + Versión de la plataforma de Fargate 1.4.0 o posterior. **Amazon Elastic Kubernetes Service (Amazon EKS) en Amazon EC2** + El sistema operativo debe ser Linux. **Amazon EKS en AWS Fargate** + Plataforma de Fargate 1.3.0 o posterior. **importante** Si KCL 3.x no puede recopilar las métricas de uso de la CPU de los procesos de trabajo, ya que no se cumplen los requisitos previos, reequilibrará la carga según el nivel de rendimiento por arrendamiento. Este mecanismo de reequilibrio alternativo garantizará que todos los procesos de trabajo obtengan niveles de rendimiento total similares en los arrendamientos asignados a cada proceso de trabajo. Para obtener más información, consulte [Cómo KCL asigna los arrendamientos a los procesos de trabajo y equilibra la carga](kcl-dynamoDB.md#kcl-assign-leases). ## Paso 6: actualizar los permisos de IAM para KCL 3.x Debe agregar los siguientes permisos al rol o la política de IAM asociada a su aplicación de consumo de KCL 3.x. Esto implica actualizar la política de IAM existente que utiliza la aplicación KCL. Para obtener más información, consulte [Permisos de IAM necesarios para las aplicaciones de consumo de KCL](kcl-iam-permissions.md). **importante** Es posible que sus aplicaciones de KCL existentes no tengan las siguientes acciones y recursos de IAM agregados en la política de IAM porque no eran necesarios en KCL 2.x. Recuerde agregarlos antes de ejecutar su aplicación de KCL 3.x: Acciones: `UpdateTable` Recursos (ARNs): `arn:aws:dynamodb:region:account:table/KCLApplicationName` Acciones: `Query` Recursos (ARNs): `arn:aws:dynamodb:region:account:table/KCLApplicationName/index/*` Acciones:`CreateTable`, `DescribeTable``Scan`,`GetItem`,`PutItem`,`UpdateItem`, `DeleteItem` Recursos (ARNs):`arn:aws:dynamodb:region:account:table/KCLApplicationName-WorkerMetricStats`, `arn:aws:dynamodb:region:account:table/KCLApplicationName-CoordinatorState` Sustituya «región», «cuenta» y «KCLApplicationnombre» por su propio Región de AWS Cuenta de AWS número y nombre de la aplicación de KCL, respectivamente. ARNs Si utiliza configuraciones para personalizar los nombres de las tablas de metadatos creadas por KCL, utilice los nombres de tabla especificados en lugar del nombre de la aplicación de KCL. ## Paso 7: implementar el código KCL 3.x para sus procesos de trabajo Tras haber establecido la configuración necesaria para la migración y haber completado todas las listas de verificación anteriores, podrá crear e implementar el código para sus procesos de trabajo. **nota** Si ve un error de compilación en el `LeaseManagementConfig` constructor, consulte Error de [compilación con el LeaseManagementConfig constructor para obtener información sobre la](https://docs.aws.amazon.com/streams/latest/dev/troubleshooting-consumers.html#compilation-error-leasemanagementconfig) solución de problemas. ## Paso 8: completar la migración Durante la implementación del código KCL 3.x, KCL sigue utilizando el algoritmo de asignación de arrendamientos de KCL 2.x. Cuando haya implementado correctamente el código KCL 3.x en todos sus procesos de trabajo, KCL lo detectará automáticamente y pasará al nuevo algoritmo de asignación de arrendamientos según la utilización de los recursos de los procesos de trabajo. Para obtener más información sobre el nuevo algoritmo de asignación de arrendamiento, consulte [Cómo KCL asigna los arrendamientos a los procesos de trabajo y equilibra la carga](kcl-dynamoDB.md#kcl-assign-leases). Durante la implementación, puede supervisar el proceso de migración con las siguientes métricas emitidas a CloudWatch. Puede supervisar las métricas de la operación `Migration`. Todas las métricas son per-KCL-application métricas y se configuran en el nivel de `SUMMARY` métrica. Si la estadística `Sum` de la métrica `CurrentState:3xWorker` coincide con el número total de procesos de trabajo de su aplicación de KCL, indica que la migración a KCL 3.x se ha completado correctamente. **importante** KCL tarda al menos 10 minutos en cambiar al nuevo algoritmo de asignación de arrendatarios una vez que todos los procesos de trabajo estén preparados para ejecutarlo. **CloudWatch métricas para el proceso de migración a KCL** | Métricas | Description (Descripción) | | --- | --- | | CurrentState:3xWorker | La cantidad de procesos de trabajo de KCL se migró correctamente a KCL 3.x y se ejecutó el nuevo algoritmo de asignación de arrendamientos. Si el recuento `Sum` de esta métrica coincide con el número total de sus procesos de trabajo, indica que la migración a KCL 3.x se ha completado correctamente. [\[See the AWS documentation website for more details\]](http://docs.aws.amazon.com/es_es/streams/latest/dev/kcl-migration-from-2-3.html) | | CurrentState:2xCompatibleWorker | El número de procesos de trabajo de KCL que utilizan el modo compatible con KCL 2.x durante el proceso de migración. Un valor distinto de cero en esta métrica indica que la migración aún está en curso. [\[See the AWS documentation website for more details\]](http://docs.aws.amazon.com/es_es/streams/latest/dev/kcl-migration-from-2-3.html) | | Fault | El número de excepciones que se encontraron durante el proceso de migración. La mayoría de estas excepciones son errores transitorios, y KCL 3.x volverá a intentar completar la migración automáticamente. Si observa un valor de métrica persistente `Fault`, revise los registros del periodo de migración para seguir solucionando problemas. Si el problema persiste, ponte en contacto con Soporte. [\[See the AWS documentation website for more details\]](http://docs.aws.amazon.com/es_es/streams/latest/dev/kcl-migration-from-2-3.html) | | GsiStatusReady | El estado de la creación del índice secundario global (GSI) en la tabla de arrendamiento. Esta métrica indica si se ha creado el GSI de la tabla de arrendamiento, un requisito previo para ejecutar KCL 3.x. El valor es 0 o 1, donde 1 indica que la creación se ha realizado correctamente. Durante un estado de reversión, esta métrica no se emitirá. Cuando vuelva a realizar la actualización, podrá reanudar la supervisión de esta métrica. [\[See the AWS documentation website for more details\]](http://docs.aws.amazon.com/es_es/streams/latest/dev/kcl-migration-from-2-3.html) | | workerMetricsReady | Emisión de métricas del estado del proceso de trabajo por parte de todos los procesos de trabajo. Las métricas indican si todos los procesos de trabajo están emitiendo métricas como el uso de la CPU. El valor es 0 o 1, donde 1 indica que todos los procesos de trabajo están emitiendo correctamente las métricas y están preparados para el nuevo algoritmo de asignación de arrendamientos. Durante un estado de reversión, esta métrica no se emitirá. Cuando vuelva a realizar la actualización, podrá reanudar la supervisión de esta métrica. [\[See the AWS documentation website for more details\]](http://docs.aws.amazon.com/es_es/streams/latest/dev/kcl-migration-from-2-3.html) | El KCL ofrece la capacidad de reversión al modo compatible con la versión 2.x durante la migración. Si la migración a KCL 3.x se realiza correctamente, recomendamos eliminar la configuración `CoordinatorConfig.clientVersionConfig` de `CLIENT_VERSION_CONFIG_COMPATIBLE_WITH_2X` si la reversión ya no es necesaria. Al eliminar esta configuración, se detiene la emisión de métricas relacionadas con la migración desde la aplicación KCL. **nota** Recomendamos supervisar el rendimiento y la estabilidad de la aplicación durante un periodo durante la migración y una vez finalizada la migración. Si observa algún problema, puede hacer que los procesos de trabajo pasen a utilizar una funcionalidad compatible con KCL 2.x mediante la [herramienta de migración de KCL](https://github.com/awslabs/amazon-kinesis-client/blob/master/amazon-kinesis-client/scripts/KclMigrationTool.py). # Restauración de la versión de KCL anterior Este tema explica los pasos para revertir al consumidor a la versión anterior. Para poder revertir, existe un proceso de dos pasos: 1. Ejecución de la [herramienta de migración de KCL](https://github.com/awslabs/amazon-kinesis-client/blob/master/amazon-kinesis-client/scripts/KclMigrationTool.py). 1. Nueva implementación del código de la versión de KCL anterior (opcional). ## Paso 1: ejecución de la herramienta de migración de KCL Cuando necesite restaurar la versión anterior de KCL, debe ejecutar la herramienta de migración de KCL. La herramienta de migración de KCL realiza dos tareas importantes: + Elimina una tabla de metadatos llamada tabla de métricas de procesos de trabajo y el índice secundario global de la tabla de arrendamiento en DynamoDB. Estos dos artefactos los crea KCL 3.x, pero no son necesarios al restaurar la versión anterior. + Hace que todos los trabajadores funcionen en un modo compatible con KCL 2.x y comiencen a utilizar el algoritmo de equilibrio de carga utilizado en las versiones anteriores de KCL. Si tiene problemas con el nuevo algoritmo de equilibrio de carga en KCL 3.x, esto mitigará el problema inmediatamente. **importante** La tabla de estados del coordinador en DynamoDB debe existir y no debe eliminarse durante el proceso de migración, restauración y avance. **nota** Es importante que todos los procesos de trabajo de la aplicación de consumo utilicen el mismo algoritmo de equilibrio de carga en un momento dado. La herramienta de migración de KCL se asegura de que todos los procesos de trabajo de la aplicación de consumo KCL 3.x cambien al modo compatible con KCL 2.x, de modo que todos los procesos de trabajo ejecuten el mismo algoritmo de equilibrio de carga durante el despliegue progresivo de su versión anterior de KCL. [Puede descargar la [herramienta de migración de KCL](https://github.com/awslabs/amazon-kinesis-client/blob/master/amazon-kinesis-client/scripts/KclMigrationTool.py) en el directorio de scripts del repositorio de KCL. GitHub](https://github.com/awslabs/amazon-kinesis-client/tree/master) Puede ejecutar el script desde cualquiera de sus procesos de trabajo o desde cualquier host que tenga los permisos necesarios para escribir en la tabla de estado de coordinadores, eliminar la tabla de métricas de los procesos de trabajo y actualizar la tabla de arrendamiento. Puede consultar [Permisos de IAM necesarios para las aplicaciones de consumo de KCL](kcl-iam-permissions.md) para conocer los permisos de IAM necesarios para ejecutar el script. Debe ejecutar el script solo una vez por aplicación de KCL. Puede ejecutar esta herramienta de migración de KCL con el siguiente comando: ``` python3 ./KclMigrationTool.py --region --mode rollback [--application_name ] [--lease_table_name ] [--coordinator_state_table_name ] [--worker_metrics_table_name ] ``` **Parámetros** + --region: reemplace `` por su. Región de AWS + --application\$1name: este parámetro es necesario si utiliza nombres predeterminados para las tablas de metadatos de DynamoDB (tabla de arrendamiento, tabla de estado de coordinadores y tabla de métricas de procesos de trabajo). Si ha especificado nombres personalizados para estas tablas, puede omitir este parámetro. Reemplace `` por el nombre de la aplicación KCL. La herramienta utiliza este nombre para derivar los nombres de tabla predeterminados si no se proporcionan nombres personalizados. + --lease\$1table\$1name (opcional): este parámetro es necesario cuando se ha establecido un nombre personalizado para la tabla de arrendamiento en la configuración de KCL. Si utiliza el nombre de tabla predeterminado, puede omitir este parámetro. Reemplace `leaseTableName` por el nombre de tabla personalizado que especificó para la tabla de arrendamiento. + --coordinator\$1state\$1table\$1name (opcional): este parámetro es necesario cuando se ha establecido un nombre personalizado para la tabla de estado de coordinadores en la configuración de KCL. Si utiliza el nombre de tabla predeterminado, puede omitir este parámetro. Reemplace `` por el nombre de tabla personalizado que especificó para la tabla de estado de coordinadores. + --worker\$1metrics\$1table\$1name (opcional): este parámetro es necesario cuando se ha establecido un nombre personalizado para la tabla de métricas de procesos de trabajo en la configuración de KCL. Si utiliza el nombre de tabla predeterminado, puede omitir este parámetro. Reemplace `` por el nombre de tabla personalizado que especificó para la tabla de métricas de procesos de trabajo. ## Paso 2: nueva implementación del código con la versión de KCL anterior (opcional) Tras ejecutar la herramienta de migración de KCL para realizar una recuperación, verá uno de los siguientes mensajes: + **Mensaje 1:** “Se completó la reversión. Su aplicación KCL ejecutaba el modo compatible con KCL 2.x. Si no observa la migración de cualquier regresión, revierta sus archivos binarios de aplicación anteriores al implementar el código con su versión de KCL anterior”. + **Acción requerida:** Esto significa que sus trabajadores estaban trabajando en el modo compatible con KCL 2.x. Si el problema persiste, vuelva a implementar el código con la versión de KCL anterior en los procesos de trabajo. + **Mensaje 2:** “Se completó la reversión. Su aplicación KCL ejecutaba el modo de funcionalidad KCL 3.x. No es necesario volver a los archivos binarios de la aplicación anterior, a menos que no vea ninguna solución al problema en 5 minutos. Si sigue teniendo problemas, regrese a sus archivos binarios de aplicación anteriores al implementar el código con la versión de KCL anterior”. + **Acción necesaria:** Esto significa que sus trabajadores estaban trabajando en el modo KCL 3.x y que la herramienta de migración de KCL cambió a todos los trabajadores al modo compatible con KCL 2.x. Si el problema se resuelve, no es necesario volver a implementar el código con la versión anterior de KCL. Si el problema persiste, vuelva a implementar el código con la versión de KCL anterior en los procesos de trabajo. # Avance a KCL 3.x después de una restauración En este tema se explican los pasos para volver a KCL 3.x después de una reversión. Cuando necesite avanzar, debe realizar un proceso de dos pasos: 1. Ejecución de la [herramienta de migración de KCL](https://github.com/awslabs/amazon-kinesis-client/blob/master/amazon-kinesis-client/scripts/KclMigrationTool.py). 1. Implemente el código con KCL 3.x. ## Paso 1: ejecución de la herramienta de migración de KCL Ejecución de la herramienta de migración de KCL. La herramienta de migración de KCL con el siguiente comando para avanzar a KCL 3.x: ``` python3 ./KclMigrationTool.py --region --mode rollforward [--application_name ] [--coordinator_state_table_name ] ``` **Parámetros** + --región: sustituya por la suya. `` Región de AWS + --application\$1name: este parámetro es obligatorio si utiliza nombres predeterminados para la tabla de estado de coordinadores. Si ha especificado nombres personalizados para la tabla de estados de coordinador, puede omitir este parámetro. Reemplace `` por el nombre de la aplicación KCL. La herramienta utiliza este nombre para derivar los nombres de tabla predeterminados si no se proporcionan nombres personalizados. + --coordinator\$1state\$1table\$1name (opcional): este parámetro es necesario cuando se ha establecido un nombre personalizado para la tabla de estado de coordinadores en la configuración de KCL. Si utiliza el nombre de tabla predeterminado, puede omitir este parámetro. Reemplace `` por el nombre de tabla personalizado que especificó para la tabla de estado de coordinadores. Después de ejecutar la herramienta de migración en modo de avance, KCL crea los siguientes recursos de DynamoDB necesarios para KCL 3.x: + Un índice secundario global en la tabla de arrendamientos + Una tabla de métricas de proceso de trabajo ## Paso 2: implementación del código con KCL 3.x Después de ejecutar la herramienta de migración de KCL para una restauración, implemente el código con KCL 3.x en los procesos de trabajo. Consulte [Paso 8: completar la migración](kcl-migration-from-2-3.md#kcl-migration-from-2-3-finish) para completar la migración. # Prácticas recomendadas para la tabla de arrendamiento con el modo de capacidad aprovisionada Si la tabla de arrendamiento de su aplicación KCL se cambió al modo de capacidad aprovisionada, KCL 3.x creará un índice secundario global en la tabla de arrendamientos con el modo de facturación aprovisionado y las mismas unidades de capacidad de lectura (RCU) y unidades de capacidad de escritura (WCU) que la tabla de arrendamiento base. Cuando se cree el índice secundario global, como recomendación, debe supervisar el uso del índice secundario global en la consola de DynamoDB y ajustar las unidades de capacidad si es necesario. Para obtener una guía más detallada sobre cómo cambiar el modo de capacidad de las tablas de metadatos de DynamoDB creadas por KCL, consulte [Modo de capacidad de DynamoDB para las tablas de metadatos creadas por KCL](kcl-dynamoDB.md#kcl-capacity-mode). **nota** Por defecto, KCL crea tablas de metadatos, como la tabla de arrendamiento, la tabla de métricas de procesos de trabajo y la tabla de estado de coordinadores, y el índice secundario global en la tabla de arrendamiento mediante el modo de capacidad bajo demanda. Recomendamos utilizar el modo de capacidad bajo demanda para ajustar automáticamente la capacidad en función de los cambios de uso. # Migración de KCL 1.x a KCL 3.x En este tema se explican las instrucciones para migrar a un consumidor de KCL 1.x a KCL 3.x. KCL 1.x utiliza clases e interfaces diferentes en comparación con KCL 2.x y KCL 3.x. Debe migrar primero el procesador de registros, el generador de procesadores de registros y las clases de procesos de trabajo al formato compatible con KCL 2.x y 3.x y, a continuación, seguir los pasos de migración de KCL 2.x a KCL 3.x. Puede actualizar directamente de KCL 1.x a KCL 3.x. + **Paso 1: migrar el procesador de grabación** Consulte la sección [Migración del procesador de registros](https://docs.aws.amazon.com/streams/latest/dev/kcl-migration.html#recrod-processor-migration) de la página [Migración de consumidores de KCL 1.x a KCL 2.x](https://docs.aws.amazon.com/streams/latest/dev/kcl-migration.html#recrod-processor-migration). + **Paso 2: Migrar la fábrica de procesadores de grabación** Consulte la sección [Migración del generador de procesadores de registros](https://docs.aws.amazon.com/streams/latest/dev/kcl-migration.html#recrod-processor-factory-migration) de la página [Migración de consumidores de KCL 1.x a KCL 2.x](https://docs.aws.amazon.com/streams/latest/dev/kcl-migration.html#recrod-processor-migration). + **Paso 3: Migrar al trabajador** Consulte la sección [Migración del proceso de trabajo](https://docs.aws.amazon.com/streams/latest/dev/kcl-migration.html#worker-migration) de la página [Migración de consumidores de KCL 1.x a KCL 2.x](https://docs.aws.amazon.com/streams/latest/dev/kcl-migration.html#recrod-processor-migration). + **Paso 4: migración de la configuración de KCL 1.x** Consulte la sección [Configuración del cliente de Amazon Kinesis](https://docs.aws.amazon.com/streams/latest/dev/kcl-migration.html#client-configuration) en la página [Migración de consumidores de KCL 1.x a KCL 2.x](https://docs.aws.amazon.com/streams/latest/dev/kcl-migration.html#recrod-processor-migration). + **Paso 5: revisión de la eliminación del tiempo de inactividad y eliminación de la configuración del cliente** Consulte las secciones [Eliminación del tiempo de inactividad](https://docs.aws.amazon.com/streams/latest/dev/kcl-migration.html#idle-time-removal) y [Eliminación de la configuración del cliente](https://docs.aws.amazon.com/streams/latest/dev/kcl-migration.html#client-configuration-removals) en la página [Migración de consumidores de KCL 1.x a KCL 2.x](https://docs.aws.amazon.com/streams/latest/dev/kcl-migration.html#recrod-processor-migration). + **Paso 6: Siga step-by-step las instrucciones de la guía de migración de KCL 2.x a KCL 3.x** Siga las instrucciones de la página [Migre de KCL 2.x a KCL 3.x](kcl-migration-from-2-3.md) para completar la migración. Si necesita revertir la versión anterior de KCL o actualizar a KCL 3.x después de una reversión, consulte [Restauración de la versión de KCL anterior](kcl-migration-rollback.md) y [Avance a KCL 3.x después de una restauración](kcl-migration-rollforward.md). **importante** No utilice las AWS SDK para Java versiones 2.27.19 a 2.27.23 con KCL 3.x. Estas versiones cuentan con un problema que provoca un error de excepción relacionado con el uso de DynamoDB por parte de KCL. Le recomendamos que utilice la AWS SDK para Java versión 2.28.0 o posterior para evitar este problema. # Documentación de la versión anterior de KCL Se han archivado los siguientes temas. Para ver la documentación actual de Kinesis Client Library, consulte [Uso de Kinesis Client Library](kcl.md). **importante** Las versiones 1.x y 2.x de la biblioteca de clientes de Amazon Kinesis (KCL) están obsoletas. La versión 1.x de KCL estará disponible el 30 de end-of-support enero de 2026. **Recomendamos** que migre las aplicaciones de KCL que utilizan la versión 1.x a la última versión de KCL antes del 30 de enero de 2026. Para encontrar la versión más reciente de KCL, consulte la página de la [biblioteca de clientes de Amazon Kinesis](https://github.com/awslabs/amazon-kinesis-client) en. GitHub Para obtener información sobre las versiones más recientes de KCL, consulte [Uso de Kinesis Client Library](kcl.md). Para obtener información sobre cómo migrar de KCL 1.x a KCL 3.x, consulte [Migración de KCL 1.x a KCL 3.x](kcl-migration-1-3.md). **Topics** + [ # Información sobre KCL 1.x y 2.x ](shared-throughput-kcl-consumers.md) + [ # Desarrollar consumidores personalizados con rendimiento compartido ](shared-throughput-consumers.md) + [ # Migrar consumidores de KCL 1.x a KCL 2.x ](kcl-migration.md) # Información sobre KCL 1.x y 2.x **importante** Las versiones 1.x y 2.x de la biblioteca de clientes de Amazon Kinesis (KCL) están obsoletas. La versión 1.x de KCL estará disponible el 30 de end-of-support enero de 2026. **Recomendamos** que migre las aplicaciones de KCL que utilizan la versión 1.x a la última versión de KCL antes del 30 de enero de 2026. Para encontrar la versión más reciente de KCL, consulte la página de la [biblioteca de clientes de Amazon Kinesis](https://github.com/awslabs/amazon-kinesis-client) en. GitHub Para obtener información sobre las versiones más recientes de KCL, consulte [Uso de Kinesis Client Library](kcl.md). Para obtener información sobre cómo migrar de KCL 1.x a KCL 3.x, consulte [Migración de KCL 1.x a KCL 3.x](kcl-migration-1-3.md). Uno de los métodos para desarrollar aplicaciones de consumo personalizadas que puedan procesar datos de flujos de datos de KDS consiste en utilizar Kinesis Client Library (KCL). **Topics** + [ ## Acerca de KCL (versiones anteriores) ](#shared-throughput-kcl-consumers-overview) + [ ## Versiones anteriores de KCL ](#shared-throughput-kcl-consumers-versions) + [ ## Conceptos de KCL (versiones anteriores) ](#shared-throughput-kcl-consumers-concepts) + [ ## Usar una tabla de arrendamiento para realizar el seguimiento de las particiones procesadas por la aplicación de consumo de KCL ](#shared-throughput-kcl-consumers-leasetable) + [ ## Procesar varios flujos de datos con el mismo KCL 2.x para aplicaciones de consumo de Java ](#shared-throughput-kcl-multistream) + [ ## Utilice la KCL con el registro de esquemas AWS Glue ](#shared-throughput-kcl-consumers-glue-schema-registry) **nota** Se recomienda actualizar a la última versión tanto KCL 1.x como KCL 2.x, según el escenario de uso. Tanto KCL 1.x como KCL 2.x se actualizan periódicamente con versiones más recientes que incluyen las últimas revisiones 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-client /releases.](https://github.com/awslabs/amazon-kinesis-client/releases) ## Acerca de KCL (versiones anteriores) KCL ayuda a consumir y procesar los datos de un flujo de datos de Kinesis, ya que se encarga de muchas de las tareas complejas asociadas a la computación distribuida. Estas incluyen equilibrar la carga entre varias instancias de aplicaciones de consumo, responder a los errores de las instancias de aplicaciones de consumo, comprobar los registros procesados y reaccionar ante la repartición. KCL se encarga de todas estas subtareas para que pueda centrar sus esfuerzos en escribir una lógica de procesamiento de registros personalizada. El KCL es diferente de los Kinesis Data APIs Streams que están disponibles en. AWS SDKs Los Kinesis Data APIs Streams le ayudan a gestionar muchos aspectos de Kinesis Data Streams, como la creación de transmisiones, la refragmentación y la creación y obtención de registros. KCL proporciona una capa de abstracción en torno a todas estas subtareas, específicamente para que pueda centrarse en la lógica de procesamiento de datos personalizada de su aplicación de consumo. Para obtener información sobre la API de Kinesis Data Streams, consulte la [referencia de la API de Amazon Kinesis](https://docs.aws.amazon.com/kinesis/latest/APIReference/Welcome.html). **importante** KCL es una biblioteca de Java. El soporte para lenguajes distintos de Java se proporciona mediante una interfaz multilingüe llamada MultiLangDaemon. Este daemon está basado en Java y se ejecuta en segundo plano cuando se utiliza un lenguaje de KCL distinto de Java. Por ejemplo, si instala el KCL para Python y escribe su aplicación de consumo completamente en Python, seguirá necesitando instalar Java en su sistema debido a la MultiLangDaemon. Además, MultiLangDaemon tiene algunos ajustes predeterminados que puede que necesites personalizar para tu caso de uso, por ejemplo, la AWS región a la que se conecta. Para obtener más información MultiLangDaemon sobre esto GitHub, consulte el [ MultiLangDaemon proyecto KCL](https://github.com/awslabs/amazon-kinesis-client/tree/v1.x/src/main/java/com/amazonaws/services/kinesis/multilang). KCL ejerce de intermediaria entre su lógica de procesamiento de registros y Kinesis Data Streams. ## Versiones anteriores de KCL Actualmente, puede utilizar cualquiera de las siguientes versiones compatibles de KCL para crear sus aplicaciones de consumo personalizadas: + **KCL 1.x** Para obtener más información, consulte [Desarrollar consumidores de KCL 1.x](developing-consumers-with-kcl.md) + **KCL 2.x** Para obtener más información, consulte [Desarrollar consumidores de KCL 2.x](developing-consumers-with-kcl-v2.md) Puede usar KCL 1.x o KCL 2.x para crear aplicaciones de consumo que utilicen un rendimiento compartido. Para obtener más información, consulte [Desarrollar consumidores personalizados con rendimiento compartido mediante KCL](custom-kcl-consumers.md). Para crear aplicaciones de consumo que utilicen un rendimiento dedicado (consumidores con distribución mejorada), solo puede utilizar KCL 2.x. Para obtener más información, consulte [Desarrollo de consumidores de distribución ramificada mejorada con rendimiento dedicado](enhanced-consumers.md). Para obtener información sobre las diferencias entre KCL 1.x y KCL 2.x e instrucciones sobre cómo migrar de KCL 1.x a KCL 2.x, consulte [Migrar consumidores de KCL 1.x a KCL 2.x](kcl-migration.md). ## Conceptos de KCL (versiones anteriores) + **Aplicación para consumidores de KCL**: una aplicación creada a medida con KCL y diseñada para leer y procesar registros de flujos de datos. + **Instancia de aplicación de consumo**: las aplicaciones de consumo de KCL suelen estar distribuidas y una o más instancias de aplicación se ejecutan simultáneamente para coordinar los fallos y equilibrar la carga de forma dinámica del procesamiento de registro de datos. + **Proceso de trabajo**: clase de alto nivel que utiliza una instancia de aplicación de consumo de KCL para empezar a procesar datos. **importante** Cada instancia de aplicación de consumo de KCL tiene un proceso de trabajo. El proceso de trabajo inicializa y supervisa diversas tareas, como la sincronización de la información sobre los arrendamientos y las particiones, el seguimiento de las asignaciones de las particiones y el procesamiento de los datos de las particiones. Un trabajador proporciona a KCL la información de configuración de la aplicación de consumo, como el nombre del flujo de datos cuyos registros de datos va a procesar la aplicación de consumo de KCL y las AWS credenciales necesarias para acceder a este flujo de datos. El proceso de trabajo también pone en marcha esa instancia específica de la aplicación de consumo de KCL para entregar los registros de datos del flujo de datos a los procesadores de registros. **importante** En KCL 1.x, esta clase se denomina **Proceso de trabajo**. [Para obtener más información (estos son los repositorios KCL de Java), consulte/.java. https://github.com/awslabs/ amazon-kinesis-client blob/v1.x/src/main/java/com/amazonaws/services/kinesis/clientlibrary/lib/worker/Worker](https://github.com/awslabs/amazon-kinesis-client/blob/v1.x/src/main/java/com/amazonaws/services/kinesis/clientlibrary/lib/worker/Worker.java) En KCL 2.x, esta clase se denomina **Programador**. El propósito del programador en KCL 2.x es idéntico al propósito del proceso de trabajo en KCL 1.x. [Para obtener más información sobre la clase Scheduler en KCL 2.x, consulte/.java. https://github.com/awslabs/ amazon-kinesis-client blob/master/amazon-kinesis-client/src/main/java/software/amazon/kinesis/coordinator/Scheduler](https://github.com/awslabs/amazon-kinesis-client/blob/master/amazon-kinesis-client/src/main/java/software/amazon/kinesis/coordinator/Scheduler.java) + **Arrendamiento**: datos que definen el enlace entre un proceso de trabajo y una partición. Las aplicaciones de consumo distribuidas de KCL utilizan los arrendamientos para dividir el procesamiento de registros de datos entre una flota de procesos de trabajo. En un momento dado, cada partición de registros de datos está vinculada a un proceso de trabajo en particular mediante un arrendamiento identificado por la variable **leaseKey**. De forma predeterminada, un trabajador puede tener uno o más contratos de arrendamiento (sujetos al valor de la variable **maxLeasesForWorker) al mismo** tiempo. **importante** Cada proceso de trabajo competirá por tener todos los arrendamientos disponibles para todas las particiones disponibles en un flujo de datos. Sin embargo, solo un proceso de trabajo podrá mantener satisfactoriamente cada arrendamiento a la vez. Por ejemplo, si tiene una instancia de aplicación de consumo A con el proceso de trabajo A que procesa un flujo de datos con 4 particiones, el proceso de trabajo A puede retener los arrendamientos de las particiones 1, 2, 3 y 4 al mismo tiempo. Sin embargo, si tiene dos instancias de aplicaciones de consumo: A y B con el proceso de trabajo A y el proceso de trabajo B, y estas instancias procesan un flujo de datos con 4 particiones, el proceso de trabajo A y el proceso de trabajo B no pueden retener el arrendamiento de la partición 1 al mismo tiempo. Un proceso de trabajo retiene el arrendamiento de una partición concreta hasta que esté listo para dejar de procesar los registros de datos de esta partición o hasta que falle. Cuando un proceso de trabajo deja de ser titular del arrendamiento, otro proceso de trabajo lo acepta y lo retiene. [Para obtener más información (estos son los repositorios KCL de Java), consulte [https://github.com/awslabs/amazon-kinesis-client/blob/v1.x/src/main/java/com/amazonaws/services/kinesis/leases/impl/Lease.java para KCL 1.x y https://github.com/awslabs/amazon-kinesis-client/.java](https://github.com/awslabs/amazon-kinesis-client/blob/v1.x/src/main/java/com/amazonaws/services/kinesis/leases/impl/Lease.java) para KCL 2.x. blob/master/amazon-kinesis-client/src/main/java/software/amazon/kinesis/leases/Lease](https://github.com/awslabs/amazon-kinesis-client/blob/master/amazon-kinesis-client/src/main/java/software/amazon/kinesis/leases/Lease.java) + **Tabla de arrendamiento**: una tabla exclusiva de Amazon DynamoDB que se utiliza para realizar un seguimiento de las particiones de un flujo de datos de KDS que los procesos de trabajo de la aplicación de consumo de KCL están arrendando y procesando. La tabla de arrendamiento debe permanecer sincronizada (dentro de un proceso de trabajo y entre todos los procesos de trabajo) con la información más reciente sobre las particiones del flujo de datos mientras se ejecuta la aplicación de consumo de KCL. Para obtener más información, consulte [Usar una tabla de arrendamiento para realizar el seguimiento de las particiones procesadas por la aplicación de consumo de KCL](#shared-throughput-kcl-consumers-leasetable). + **Procesador de registros**: lógica que define la forma en que su aplicación de consumo de KCL procesa los datos que obtiene de los flujos de datos. En tiempo de ejecución, una instancia de una aplicación de consumo de KCL crea una instancia de un proceso de trabajo, y este proceso de trabajo crea una instancia de un procesador de registros por cada partición que tiene en arrendamiento. ## Usar una tabla de arrendamiento para realizar el seguimiento de las particiones procesadas por la aplicación de consumo de KCL **Topics** + [ ### Qué es una tabla de arrendamiento ](#shared-throughput-kcl-consumers-what-is-leasetable) + [ ### Rendimiento ](#shared-throughput-kcl-leasetable-throughput) + [ ### Cómo se sincroniza una tabla de arrendamiento con las particiones de Kinesis Data Streams ](#shared-throughput-kcl-consumers-leasetable-sync) ### Qué es una tabla de arrendamiento Para cada aplicación de Amazon Kinesis Data Streams, KCL utiliza una tabla de arrendamiento única (almacenada en una tabla de Amazon DynamoDB) para realizar un seguimiento de las particiones de un flujo de datos de KDS que los procesos de trabajo de la aplicación de consumo de KCL están arrendando y procesando. **importante** KCL utiliza el nombre de la aplicación de consumo para crear el nombre de la tabla de arrendamiento que utiliza esta aplicación de consumo, por lo que el nombre de cada aplicación de consumo debe ser único. Puede consultar la tabla con la [consola de Amazon DynamoDB](https://docs.aws.amazon.com/amazondynamodb/latest/developerguide/ConsoleDynamoDB.html) mientras se ejecuta la aplicación de consumo. Si la tabla de arrendamiento de la aplicación de consumo de KCL no existe cuando se inicia la aplicación, uno de los procesos de trabajo crea la tabla de arrendamiento para esta aplicación. **importante** Se le realizará el cobro de los costos de su cuenta asociados a la tabla de DynamoDB, además de los costos propios asociados a Kinesis Data Streams. Cada fila de la tabla de arrendamiento representa una partición que procesan los procesos de trabajo de la aplicación de consumo. Si la aplicación de consumo de KCL procesa solo un flujo de datos, la `leaseKey` que es la clave hash de la tabla de arrendamiento será el identificador de la partición. Si es [Procesar varios flujos de datos con el mismo KCL 2.x para aplicaciones de consumo de Java](#shared-throughput-kcl-multistream), la estructura de leaseKey tendrá el siguiente aspecto: `account-id:StreamName:streamCreationTimestamp:ShardId`. Por ejemplo, `111111111:multiStreamTest-1:12345:shardId-000000000336`. Además de la ID del fragmento, cada fila incluye también los siguientes datos: + **checkpoint:** el número secuencial de punto de comprobación más reciente del fragmento. Este valor es único en todas las particiones del flujo de datos. + **checkpointSubSequenceNúmero:** cuando se utiliza la función de agregación de la biblioteca de productores de Kinesis, se trata de una extensión del **punto de control** que rastrea los registros de los usuarios individuales dentro del registro de Kinesis. + **leaseCounter:** se utiliza para el control de versiones de las asignaciones, de modo que los procesos de trabajo puedan detectar que su asignación ha sido utilizada por otro proceso de trabajo. + **leaseKey:** un identificador único para una asignación. Cada arrendamiento es específico de una partición del flujo de datos y solo lo retiene un proceso de trabajo cada vez. + **leaseOwner:** el proceso de trabajo que tiene esta asignación. + **ownerSwitchesSincePunto de control:** cuántas veces este contrato de arrendamiento ha cambiado de trabajadores desde la última vez que se emitió un punto de control. + **parentShardId:** Se utiliza para garantizar que el fragmento principal se procese por completo antes de que comience el procesamiento en los fragmentos secundarios. Así, se garantiza que los registros se procesen en el mismo orden en el que se introdujeron en la secuencia. + **hashrange:** lo utiliza `PeriodicShardSyncManager` para ejecutar sincronizaciones periódicas para encontrar las particiones que faltan en la tabla de arrendamiento y crear arrendamientos para ellas si es necesario. **nota** Estos datos están presentes en la tabla de arrendamiento de todas las particiones a partir de KCL 1.14 y KCL 2.3. Para obtener más información sobre `PeriodicShardSyncManager` y la sincronización periódica entre los arrendamientos y las particiones, consulte [Cómo se sincroniza una tabla de arrendamiento con las particiones de Kinesis Data Streams](#shared-throughput-kcl-consumers-leasetable-sync). + **childshards:** lo utiliza `LeaseCleanupManager` para revisar el estado de procesamiento de la partición secundaria y decidir si la partición principal se puede eliminar de la tabla de arrendamiento. **nota** Estos datos están presentes en la tabla de arrendamiento de todas las particiones a partir de KCL 1.14 y KCL 2.3. + **shardID:** ID de la partición. **nota** Estos datos solo están presentes en la tabla de arrendamiento si es [Procesar varios flujos de datos con el mismo KCL 2.x para aplicaciones de consumo de Java](#shared-throughput-kcl-multistream). Esto solo se admite en KCL 2.x para Java, a partir de KCL 2.3 para Java y versiones posteriores. + **nombre del flujo** El identificador del flujo de datos en el siguiente formato: `account-id:StreamName:streamCreationTimestamp`. **nota** Estos datos solo están presentes en la tabla de arrendamiento si se dedica al [Procesar varios flujos de datos con el mismo KCL 2.x para aplicaciones de consumo de Java](#shared-throughput-kcl-multistream). Esto solo se admite en KCL 2.x para Java, a partir de KCL 2.3 para Java y versiones posteriores. ### Rendimiento Si la aplicación de Amazon Kinesis Data Streams recibe excepciones de rendimiento aprovisionado, debe aumentar el rendimiento aprovisionado para la tabla de DynamoDB. KCL crea la tabla con un rendimiento aprovisionado de 10 lecturas por segundo y 10 escrituras por segundo, pero esto podría no ser suficiente para su aplicación. Por ejemplo, si su aplicación de Amazon Kinesis Data Streams crea frecuentemente puntos de comprobación u opera en un flujo que se compone de muchas particiones, es posible que necesite más rendimiento. Para obtener información sobre el rendimiento aprovisionado en DynamoDB, consulte [Modo de capacidad de lectura y escritura](https://docs.aws.amazon.com/amazondynamodb/latest/developerguide/HowItWorks.ReadWriteCapacityMode.html) y [Uso de tablas y datos](https://docs.aws.amazon.com/amazondynamodb/latest/developerguide/WorkingWithDDTables.html) en la *Guía para desarrolladores de Amazon DynamoDB*. ### Cómo se sincroniza una tabla de arrendamiento con las particiones de Kinesis Data Streams Los procesos de trabajo de las aplicaciones de consumo de KCL utilizan los arrendamientos para procesar particiones de un flujo de datos determinado. La información sobre qué proceso de trabajo está arrendando cada partición en un momento dado se almacena en una tabla de arrendamiento. La tabla de arrendamiento debe permanecer sincronizada con la información más reciente sobre la partición del flujo de datos mientras se ejecuta la aplicación de consumo de KCL. KCL sincroniza la tabla de arrendamiento con la información de las particiones obtenida del servicio Kinesis Data Streams durante el arranque de la aplicación de consumo (ya sea al inicializar o reiniciar la aplicación de consumo) y también siempre que una partición que se esté procesando llegue a su fin (repartición). En otras palabras, los procesos de trabajo o una aplicación de consumo de KCL se sincronizan con el flujo de datos que están procesando durante el arranque inicial de la aplicación de consumo y siempre que la aplicación de consumo encuentra un evento de repartición del flujo de datos. **Topics** + [ #### Sincronización en KCL 1.0 a 1.13 y KCL 2.0 a 2.2 ](#shared-throughput-kcl-consumers-leasetable-sync-old) + [ #### Sincronización en KCL 2.x, a partir de KCL 2.3 y versiones posteriores ](#shared-throughput-kcl-consumers-leasetable-sync-new-kcl2) + [ #### Sincronización en KCL 1.x, a partir de KCL 1.14 y versiones posteriores ](#shared-throughput-kcl-consumers-leasetable-sync-new-kcl1) #### Sincronización en KCL 1.0 a 1.13 y KCL 2.0 a 2.2 En KCL 1.0 a 1.13 y KCL 2.0 a 2.2, durante el arranque de la aplicación de consumo y también durante cada evento de refragmentación del flujo de datos, KCL sincroniza la tabla de arrendamiento con la información de los fragmentos adquirida en el servicio Kinesis Data Streams invocando la o la detección. `ListShards` `DescribeStream` APIs En todas las versiones de KCL enumeradas anteriormente, cada empleado de una aplicación de consumo de KCL completa los siguientes pasos para realizar el proceso de lease/shard sincronización durante el arranque de la aplicación de consumo y en cada evento de refragmentación de la transmisión: + Obtiene todos las particiones de datos del flujo que se está procesando. + Obtiene todos los arrendamientos de particiones de la tabla de arrendamiento. + Filtra cada partición abierta que no tenga arrendamientos en la tabla de arrendamiento. + Repite todas las particiones abiertas encontradas y para cada partición abierta sin un elemento principal abierto: + Recorre el árbol jerárquico siguiendo la ruta de sus antecesores para determinar si la partición es descendiente. Una partición se considera descendiente si se está procesando una partición anterior (en la tabla de arrendamiento se indica el arrendamiento de la partición anterior) o si se debe procesar una partición anterior (por ejemplo, si la posición inicial es `TRIM_HORIZON` o `AT_TIMESTAMP`). + Si la partición abierta en el contexto es descendiente, KCL comprueba la partición en función de su posición inicial y crea arrendamientos para sus elementos principales, si es necesario. #### Sincronización en KCL 2.x, a partir de KCL 2.3 y versiones posteriores A partir de las últimas versiones compatibles de KCL 2.x (KCL 2.3) y versiones posteriores, la biblioteca ahora admite los siguientes cambios en el proceso de sincronización. Estos cambios de lease/shard sincronización reducen considerablemente el número de llamadas a la API que realizan las aplicaciones de consumo de KCL al servicio Kinesis Data Streams y optimizan la administración de arrendamientos en su aplicación de consumo de KCL. + Durante el arranque de la aplicación, si la tabla de arrendamiento está vacía, KCL utiliza la opción de filtrado de la API `ListShard` (el parámetro de solicitud `ShardFilter` opcional) para recuperar y crear arrendamientos únicamente para una instantánea de las particiones abiertas en el momento especificado por el parámetro `ShardFilter`. El parámetro `ShardFilter` permite filtrar la respuesta de la API `ListShards`. La única propiedad obligatoria del parámetro `ShardFilter` es `Type`. KCL utiliza la propiedad de filtro `Type` y los siguientes valores válidos para identificar y devolver una instantánea de las particiones abiertas que podrían requerir nuevos arrendamientos: + `AT_TRIM_HORIZON`: la respuesta incluye todas las particiones que estaban abiertas en `TRIM_HORIZON`. + `AT_LATEST`: la respuesta incluye solo las particiones actualmente abiertas del flujo de datos. + `AT_TIMESTAMP`: la respuesta incluye todas las particiones cuya marca de tiempo de inicio es anterior o igual a la marca de tiempo dada y cuya marca de tiempo de finalización es posterior o igual que la marca de tiempo dada, o que aún están abiertas. `ShardFilter` se utiliza al crear arrendamientos para una tabla de arrendamiento vacía con el fin de inicializar los arrendamientos de una instantánea de las particiones especificadas en `RetrievalConfig#initialPositionInStreamExtended`. Para obtener más información acerca de `ShardFilter`, consulte [https://docs.aws.amazon.com/kinesis/latest/APIReference/API_ShardFilter.html](https://docs.aws.amazon.com/kinesis/latest/APIReference/API_ShardFilter.html). + En lugar de que todos los trabajadores realicen la lease/shard sincronización para mantener la tabla de arrendamientos actualizada con los fragmentos más recientes del flujo de datos, un único líder obrero elegido realiza la sincronización entre el arrendamiento y el fragmento. + KCL 2.3 utiliza el parámetro de `ChildShards` retorno de `GetRecords` y el `SubscribeToShard` APIs para realizar la lease/shard sincronización que se produce en los fragmentos cerrados, lo que permite a `SHARD_END` un trabajador de KCL crear únicamente arrendamientos para los fragmentos secundarios del fragmento que ha terminado de procesar. Para compartirla en todas las aplicaciones de consumo, esta optimización de la lease/shard sincronización utiliza el parámetro de la API. `ChildShards` `GetRecords` En el caso de las aplicaciones de consumo dedicadas al rendimiento (distribución mejorada), esta optimización de la lease/shard sincronización utiliza el `ChildShards` parámetro de la `SubscribeToShard` API. Para obtener más información, consulte [GetRecords](https://docs.aws.amazon.com/kinesis/latest/APIReference/API_GetRecords.html), [SubscribeToShards](https://docs.aws.amazon.com/kinesis/latest/APIReference/API_SubscribeToShard.html) y [ChildShard](https://docs.aws.amazon.com/kinesis/latest/APIReference/API_ChildShard.html). + Con los cambios anteriores, el comportamiento de KCL está pasando del modelo en el que todos los procesos de trabajo aprenden sobre todas las particiones existentes al modelo en el que los procesos de trabajo aprenden solo sobre las particiones secundarias de las particiones que son propiedad de cada proceso de trabajo. Por lo tanto, además de la sincronización que se produce durante el arranque de las aplicaciones de consumo y los eventos de refragmentación, KCL ahora también realiza shard/lease escaneos periódicos adicionales para identificar cualquier posible laguna en la tabla de arrendamientos (en otras palabras, para obtener información sobre todos los fragmentos nuevos) a fin de garantizar que se procese todo el rango de hash del flujo de datos y crear arrendamientos para ellos si es necesario. `PeriodicShardSyncManager`es el componente responsable de ejecutar escaneos periódicos. lease/shard Para obtener más información sobre `PeriodicShardSyncManager` KCL 2.3, consulte [https://github.com/awslabs/amazon-kinesis-client/blob/master/amazon-kinesis-client/src/main/java/software/amazon/kinesis/leases/LeaseManagementConfig.java \$1L201](https://github.com/awslabs/amazon-kinesis-client/blob/master/amazon-kinesis-client/src/main/java/software/amazon/kinesis/leases/LeaseManagementConfig.java#L201-L213) -L213. En KCL 2.3, hay nuevas opciones de configuración disponibles para configurar `PeriodicShardSyncManager` en `LeaseManagementConfig`: [\[See the AWS documentation website for more details\]](http://docs.aws.amazon.com/es_es/streams/latest/dev/shared-throughput-kcl-consumers.html) Ahora también se emiten nuevas CloudWatch métricas para monitorear el estado del. `PeriodicShardSyncManager` Para obtener más información, consulte [PeriodicShardSyncManager](monitoring-with-kcl.md#periodic-task). + Incluye una optimización de `HierarchicalShardSyncer` para crear solo arrendamientos para una capa de particiones. #### Sincronización en KCL 1.x, a partir de KCL 1.14 y versiones posteriores A partir de las últimas versiones compatibles de KCL 1.x (KCL 1.14) y versiones posteriores, la biblioteca ahora admite los siguientes cambios en el proceso de sincronización. Estos cambios de lease/shard sincronización reducen considerablemente el número de llamadas a la API que realizan las aplicaciones de consumo de KCL al servicio Kinesis Data Streams y optimizan la administración de arrendamientos en su aplicación de consumo de KCL. + Durante el arranque de la aplicación, si la tabla de arrendamiento está vacía, KCL utiliza la opción de filtrado de la API `ListShard` (el parámetro de solicitud `ShardFilter` opcional) para recuperar y crear arrendamientos únicamente para una instantánea de las particiones abiertas en el momento especificado por el parámetro `ShardFilter`. El parámetro `ShardFilter` permite filtrar la respuesta de la API `ListShards`. La única propiedad obligatoria del parámetro `ShardFilter` es `Type`. KCL utiliza la propiedad de filtro `Type` y los siguientes valores válidos para identificar y devolver una instantánea de las particiones abiertas que podrían requerir nuevos arrendamientos: + `AT_TRIM_HORIZON`: la respuesta incluye todas las particiones que estaban abiertas en `TRIM_HORIZON`. + `AT_LATEST`: la respuesta incluye solo las particiones actualmente abiertas del flujo de datos. + `AT_TIMESTAMP`: la respuesta incluye todas las particiones cuya marca de tiempo de inicio es anterior o igual a la marca de tiempo dada y cuya marca de tiempo de finalización es posterior o igual que la marca de tiempo dada, o que aún están abiertas. `ShardFilter` se utiliza al crear arrendamientos para una tabla de arrendamiento vacía con el fin de inicializar los arrendamientos de una instantánea de las particiones especificadas en `KinesisClientLibConfiguration#initialPositionInStreamExtended`. Para obtener más información acerca de `ShardFilter`, consulte [https://docs.aws.amazon.com/kinesis/latest/APIReference/API_ShardFilter.html](https://docs.aws.amazon.com/kinesis/latest/APIReference/API_ShardFilter.html). + En lugar de que todos los trabajadores realicen la lease/shard sincronización para mantener la tabla de arrendamientos actualizada con los fragmentos más recientes del flujo de datos, un único líder obrero elegido realiza la sincronización entre el arrendamiento y el fragmento. + KCL 1.14 utiliza el parámetro de `ChildShards` retorno de `GetRecords` y el `SubscribeToShard` APIs para realizar la lease/shard sincronización que se produce en los fragmentos cerrados, lo que permite a `SHARD_END` un trabajador de KCL crear únicamente arrendamientos para los fragmentos secundarios del fragmento que ha terminado de procesar. Para obtener más información, consulte [GetRecords](https://docs.aws.amazon.com/kinesis/latest/APIReference/API_GetRecords.html) y [ChildShard](https://docs.aws.amazon.com/kinesis/latest/APIReference/API_ChildShard.html). + Con los cambios anteriores, el comportamiento de KCL está pasando del modelo en el que todos los procesos de trabajo aprenden sobre todas las particiones existentes al modelo en el que los procesos de trabajo aprenden solo sobre las particiones secundarias de las particiones que son propiedad de cada proceso de trabajo. Por lo tanto, además de la sincronización que se produce durante el arranque de las aplicaciones de consumo y los eventos de refragmentación, KCL ahora también realiza shard/lease escaneos periódicos adicionales para identificar cualquier posible laguna en la tabla de arrendamientos (en otras palabras, para obtener información sobre todos los fragmentos nuevos) a fin de garantizar que se procese todo el rango de hash del flujo de datos y crear arrendamientos para ellos si es necesario. `PeriodicShardSyncManager`es el componente responsable de ejecutar escaneos periódicos. lease/shard Cuando `KinesisClientLibConfiguration#shardSyncStrategyType` está establecido en `ShardSyncStrategyType.SHARD_END`, `PeriodicShardSync leasesRecoveryAuditorInconsistencyConfidenceThreshold` se utiliza para determinar el umbral del número de exámenes consecutivos que contienen lagunas en la tabla de arrendamiento, tras lo cual se exige la sincronización de las particiones. Cuando `KinesisClientLibConfiguration#shardSyncStrategyType` se establece en `ShardSyncStrategyType.PERIODIC`, `leasesRecoveryAuditorInconsistencyConfidenceThreshold` se ignora. Para obtener más información sobre `PeriodicShardSyncManager` KCL 1.14, consulte [https://github.com/awslabs/amazon-kinesis-client/blob/v1.x/src/main/java/com/amazonaws/services/kinesis/clientlibrary/lib/worker/KinesisClientLibConfiguration.java](https://github.com/awslabs/amazon-kinesis-client/blob/v1.x/src/main/java/com/amazonaws/services/kinesis/clientlibrary/lib/worker/KinesisClientLibConfiguration.java#L987-L999) \$1L987 -L999. En KCL 1.14, hay una nueva opción de configuración disponible para configurar `PeriodicShardSyncManager` en `LeaseManagementConfig`: [\[See the AWS documentation website for more details\]](http://docs.aws.amazon.com/es_es/streams/latest/dev/shared-throughput-kcl-consumers.html) Ahora también se emiten nuevas CloudWatch métricas para monitorear el estado del. `PeriodicShardSyncManager` Para obtener más información, consulte [PeriodicShardSyncManager](monitoring-with-kcl.md#periodic-task). + KCL 1.14 ahora también admite la limpieza de arrendamientos diferidos. `LeaseCleanupManager` elimina los arrendamientos de forma asíncrona al llegar a `SHARD_END`, cuando una partición ha caducado pasado el periodo de retención del flujo de datos o se ha cerrado como resultado de una operación de repartición. Están disponibles nuevas opciones de configuración para `LeaseCleanupManager`. [\[See the AWS documentation website for more details\]](http://docs.aws.amazon.com/es_es/streams/latest/dev/shared-throughput-kcl-consumers.html) + Incluye una optimización de `KinesisShardSyncer` para crear solo arrendamientos para una capa de particiones. ## Procesar varios flujos de datos con el mismo KCL 2.x para aplicaciones de consumo de Java En esta sección se describen los siguientes cambios en KCL 2.x para Java, que permiten crear aplicaciones de consumo de KCL que pueden procesar más de un flujo de datos al mismo tiempo. **importante** El procesamiento de varios flujos solo se admite en KCL 2.x para Java, a partir de KCL 2.3 para Java y versiones posteriores. El procesamiento de varios flujos NO es compatible con ningún otro lenguaje en el que se pueda implementar KCL 2.x. El procesamiento de varios flujos NO es compatible con ninguna versión de KCL 1.x. + **MultistreamTracker interfaz** Para crear una aplicación de consumo que pueda procesar múltiples transmisiones al mismo tiempo, debe implementar una nueva interfaz llamada [MultistreamTracker](https://github.com/awslabs/amazon-kinesis-client/blob/0c5042dadf794fe988438436252a5a8fe70b6b0b/amazon-kinesis-client/src/main/java/software/amazon/kinesis/processor/MultiStreamTracker.java). Esta interfaz incluye el método `streamConfigList` que devuelve la lista de flujos de datos y sus configuraciones para que los procese la aplicación de consumo de KCL. Tenga en cuenta que los flujos de datos que se procesan pueden cambiar durante el tiempo de ejecución de la aplicación de consumo. KCL llama a `streamConfigList` periódicamente para obtener información sobre los cambios en los flujos de datos que se van a procesar. El `streamConfigList` método rellena la [StreamConfig](https://github.com/awslabs/amazon-kinesis-client/blob/0c5042dadf794fe988438436252a5a8fe70b6b0b/amazon-kinesis-client/src/main/java/software/amazon/kinesis/common/StreamConfig.java#L23)lista. ``` package software.amazon.kinesis.common; import lombok.Data; import lombok.experimental.Accessors; @Data @Accessors(fluent = true) public class StreamConfig { private final StreamIdentifier streamIdentifier; private final InitialPositionInStreamExtended initialPositionInStreamExtended; private String consumerArn; } ``` Tenga en cuenta que los campos `StreamIdentifier` y `InitialPositionInStreamExtended` son obligatorios, aunque `consumerArn` es opcional. Debe proporcionar `consumerArn` únicamente si utiliza KCL 2.x para implementar una aplicación de consumo con distribución mejorada. Para obtener más información al respecto`StreamIdentifier`, consulte [https://github.com/awslabs/amazon-kinesis-client/blob/v2.5.8/amazon-kinesis-client/src/main/java/software/amazon/kinesis/common/StreamIdentifier.java \$1L129.](https://github.com/awslabs/amazon-kinesis-client/blob/v2.5.8/amazon-kinesis-client/src/main/java/software/amazon/kinesis/common/StreamIdentifier.java#L129) Para crear un `StreamIdentifier`, le recomendamos que cree una instancia multiflujo a partir de `streamArn` y la `streamCreationEpoch` que esté disponible en la versión 2.5.0 y versiones posteriores. En las versiones KCL 2.3 y 2.4, que no son compatibles con `streamArm`, cree una instancia multiflujo con el formato `account-id:StreamName:streamCreationTimestamp`. Este formato quedará obsoleto y dejará de ser compatible a partir de la próxima versión principal. `MultistreamTracker` también incluye una estrategia para eliminar los arrendamientos de flujos antiguos en la tabla de arrendamiento (`formerStreamsLeasesDeletionStrategy`). Tenga en cuenta que la estrategia NO SE PUEDE cambiar durante el tiempo de ejecución de la aplicación de consumo. Para obtener más información, consulte [https://github.com/awslabs/amazon-kinesis-clientb/amazon-kinesis-client/src/main/java/software/amazon/kinesis/processor/FormerStreamsLeasesDeletionStrategy/blob/0c5042dadf794fe988438436252a5a8fe70b6b0](https://github.com/awslabs/amazon-kinesis-client/blob/0c5042dadf794fe988438436252a5a8fe70b6b0b/amazon-kinesis-client/src/main/java/software/amazon/kinesis/processor/FormerStreamsLeasesDeletionStrategy.java) .java + [ConfigsBuilder](https://github.com/awslabs/amazon-kinesis-client/blob/0c5042dadf794fe988438436252a5a8fe70b6b0b/amazon-kinesis-client/src/main/java/software/amazon/kinesis/common/ConfigsBuilder.java)es una clase que abarca toda la aplicación y que puede utilizar para especificar todos los ajustes de configuración de KCL 2.x que se utilizarán al crear su aplicación de consumo de KCL. `ConfigsBuilder`la clase ahora es compatible con la interfaz. `MultistreamTracker` Puede inicializar ConfigsBuilder cualquiera de las dos con el nombre del flujo de datos desde el que se van a consumir los registros: ``` /** * Constructor to initialize ConfigsBuilder with StreamName * @param streamName * @param applicationName * @param kinesisClient * @param dynamoDBClient * @param cloudWatchClient * @param workerIdentifier * @param shardRecordProcessorFactory */ public ConfigsBuilder(@NonNull String streamName, @NonNull String applicationName, @NonNull KinesisAsyncClient kinesisClient, @NonNull DynamoDbAsyncClient dynamoDBClient, @NonNull CloudWatchAsyncClient cloudWatchClient, @NonNull String workerIdentifier, @NonNull ShardRecordProcessorFactory shardRecordProcessorFactory) { this.appStreamTracker = Either.right(streamName); this.applicationName = applicationName; this.kinesisClient = kinesisClient; this.dynamoDBClient = dynamoDBClient; this.cloudWatchClient = cloudWatchClient; this.workerIdentifier = workerIdentifier; this.shardRecordProcessorFactory = shardRecordProcessorFactory; } ``` O bien, puede inicializarlo ConfigsBuilder `MultiStreamTracker` si desea implementar una aplicación de consumo de KCL que procese varios flujos al mismo tiempo. ``` * Constructor to initialize ConfigsBuilder with MultiStreamTracker * @param multiStreamTracker * @param applicationName * @param kinesisClient * @param dynamoDBClient * @param cloudWatchClient * @param workerIdentifier * @param shardRecordProcessorFactory */ public ConfigsBuilder(@NonNull MultiStreamTracker multiStreamTracker, @NonNull String applicationName, @NonNull KinesisAsyncClient kinesisClient, @NonNull DynamoDbAsyncClient dynamoDBClient, @NonNull CloudWatchAsyncClient cloudWatchClient, @NonNull String workerIdentifier, @NonNull ShardRecordProcessorFactory shardRecordProcessorFactory) { this.appStreamTracker = Either.left(multiStreamTracker); this.applicationName = applicationName; this.kinesisClient = kinesisClient; this.dynamoDBClient = dynamoDBClient; this.cloudWatchClient = cloudWatchClient; this.workerIdentifier = workerIdentifier; this.shardRecordProcessorFactory = shardRecordProcessorFactory; } ``` + Con la compatibilidad con varios flujos implementada para la aplicación de consumo de KCL, cada fila de la tabla de arrendamiento de la aplicación ahora contiene el ID de la partición y el nombre del flujo de los varios flujos de datos que procesa esta aplicación. + Cuando se implementa la compatibilidad con varios flujos para la aplicación de consumo de KCL, leaseKey adopta la siguiente estructura: `account-id:StreamName:streamCreationTimestamp:ShardId`. Por ejemplo, `111111111:multiStreamTest-1:12345:shardId-000000000336`. **importante** Cuando la aplicación de consumo de KCL existente está configurada para procesar solo un flujo de datos, leaseKey (que es la clave hash de la tabla de arrendamiento) es el ID de la partición. Si vuelve a configurar esta aplicación de consumo de KCL existente para procesar varios flujos de datos, se rompe la tabla de arrendamiento, ya que con la compatibilidad con varios flujos, la estructura de leaseKey debe ser la siguiente: `account-id:StreamName:StreamCreationTimestamp:ShardId`. ## Utilice la KCL con el registro de esquemas AWS Glue 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 a través de KCL en Java. **importante** Actualmente, la integración de Kinesis Data Streams AWS Glue y Schema Registry solo se admite para las transmisiones de datos de Kinesis que utilizan consumidores de KCL 2.3 implementados en Java. No se proporciona soporte multilingüe. No se admiten las aplicaciones de consumo de KCL 1.0. No se admiten las aplicaciones de consumo de KCL 2.x anteriores a KCL 2.3. Para obtener instrucciones detalladas sobre cómo configurar la integración de Kinesis Data Streams con Schema Registry mediante la KCL, 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. # Desarrollar consumidores personalizados con rendimiento compartido **importante** Las versiones 1.x y 2.x de la biblioteca de clientes de Amazon Kinesis (KCL) están obsoletas. La versión 1.x de KCL estará disponible el 30 de enero de end-of-support 2026. **Recomendamos** que migre las aplicaciones de KCL que utilizan la versión 1.x a la última versión de KCL antes del 30 de enero de 2026. Para encontrar la versión más reciente de KCL, consulte la página de la [biblioteca de clientes de Amazon Kinesis](https://github.com/awslabs/amazon-kinesis-client) en. GitHub Para obtener información sobre las versiones más recientes de KCL, consulte [Uso de Kinesis Client Library](kcl.md). Para obtener información sobre cómo migrar de KCL 1.x a KCL 3.x, consulte [Migración de KCL 1.x a KCL 3.x](kcl-migration-1-3.md). Si no necesita un rendimiento específico al recibir datos de Kinesis Data Streams y tampoco necesita demoras de propagación de lectura inferiores a 200 ms, puede crear aplicaciones consumidoras tal y como se describe en los temas siguientes. Puede utilizar Kinesis Client Library (KCL) o Kinesis Client Library (KCL) o AWS SDK para Java. **Topics** + [ # Desarrollar consumidores personalizados con rendimiento compartido mediante KCL ](custom-kcl-consumers.md) Para obtener más información sobre cómo crear consumidores que puedan recibir registros de flujos de datos de Kinesis con un rendimiento específico, consulte [Desarrollo de consumidores de distribución ramificada mejorada con rendimiento dedicado](enhanced-consumers.md). # Desarrollar consumidores personalizados con rendimiento compartido mediante KCL **importante** Las versiones 1.x y 2.x de la biblioteca de clientes de Amazon Kinesis (KCL) están obsoletas. La versión 1.x de KCL estará disponible el 30 de end-of-support enero de 2026. **Recomendamos** que migre las aplicaciones de KCL que utilizan la versión 1.x a la última versión de KCL antes del 30 de enero de 2026. Para encontrar la versión más reciente de KCL, consulte la página de la [biblioteca de clientes de Amazon Kinesis](https://github.com/awslabs/amazon-kinesis-client) en. GitHub Para obtener información sobre las versiones más recientes de KCL, consulte [Uso de Kinesis Client Library](kcl.md). Para obtener información sobre cómo migrar de KCL 1.x a KCL 3.x, consulte [Migración de KCL 1.x a KCL 3.x](kcl-migration-1-3.md). Uno de los métodos para desarrollar una aplicación de consumo personalizada con un rendimiento compartido consiste en utilizar Kinesis Client Library (KCL). Elegir uno de los siguientes temas para la versión de KCL que utilice. **Topics** + [ # Desarrollar consumidores de KCL 1.x ](developing-consumers-with-kcl.md) + [ # Desarrollar consumidores de KCL 2.x ](developing-consumers-with-kcl-v2.md) # Desarrollar consumidores de KCL 1.x **importante** Las versiones 1.x y 2.x de la biblioteca de clientes de Amazon Kinesis (KCL) están obsoletas. La versión 1.x de KCL estará disponible el 30 de end-of-support enero de 2026. **Recomendamos** que migre las aplicaciones de KCL que utilizan la versión 1.x a la última versión de KCL antes del 30 de enero de 2026. Para encontrar la versión más reciente de KCL, consulte la página de la [biblioteca de clientes de Amazon Kinesis](https://github.com/awslabs/amazon-kinesis-client) en. GitHub Para obtener información sobre las versiones más recientes de KCL, consulte [Uso de Kinesis Client Library](kcl.md). Para obtener información sobre cómo migrar de KCL 1.x a KCL 3.x, consulte [Migración de KCL 1.x a KCL 3.x](kcl-migration-1-3.md). Puede desarrollar una aplicación de consumo para Amazon Kinesis Data Streams mediante Kinesis Client Library (KCL). Para obtener más información acerca de, consulte [Acerca de KCL (versiones anteriores)](shared-throughput-kcl-consumers.md#shared-throughput-kcl-consumers-overview). Elija uno de los siguientes temas según la opción que desee utilizar. **Topics** + [ # Desarrollar un consumidor de Kinesis Client Library en Java ](kinesis-record-processor-implementation-app-java.md) + [ # Desarrollar un consumidor de Kinesis Client Library en Node.js ](kinesis-record-processor-implementation-app-nodejs.md) + [ # Desarrollar un consumidor de Kinesis Client Library en .NET ](kinesis-record-processor-implementation-app-dotnet.md) + [ # Desarrollar un consumidor de Kinesis Client Library en Python ](kinesis-record-processor-implementation-app-py.md) + [ # Desarrollar un consumidor de Kinesis Client Library en Ruby ](kinesis-record-processor-implementation-app-ruby.md) # Desarrollar un consumidor de Kinesis Client Library en Java **importante** Las versiones 1.x y 2.x de la biblioteca de clientes de Amazon Kinesis (KCL) están obsoletas. La versión 1.x de KCL estará disponible el 30 de end-of-support enero de 2026. **Recomendamos** que migre las aplicaciones de KCL que utilizan la versión 1.x a la última versión de KCL antes del 30 de enero de 2026. Para encontrar la versión más reciente de KCL, consulte la página de la [biblioteca de clientes de Amazon Kinesis](https://github.com/awslabs/amazon-kinesis-client) en. GitHub Para obtener información sobre las versiones más recientes de KCL, consulte [Uso de Kinesis Client Library](kcl.md). Para obtener información sobre cómo migrar de KCL 1.x a KCL 3.x, consulte [Migración de KCL 1.x a KCL 3.x](kcl-migration-1-3.md). Puede utilizar Kinesis Client Library (KCL) para crear aplicaciones que procesen datos de los flujos de datos de Kinesis. Kinesis Client Library está disponible en varios idiomas. En este tema se habla de Java. Para ver la referencia sobre Javadoc, consulte el tema sobre [AWS Javadoc](https://docs.aws.amazon.com/AWSJavaSDK/latest/javadoc/com/amazonaws/services/kinesis/AmazonKinesisClient.html) de Class. AmazonKinesisClient Para descargar el KCL de Java GitHub, vaya a la [biblioteca de clientes de Kinesis (](https://github.com/awslabs/amazon-kinesis-client)Java). Para localizar KCL para Java en Apache Maven, vaya a la página de [resultados de la búsqueda de KCL](https://search.maven.org/#search|ga|1|amazon-kinesis-client). Para descargar un código de muestra para una aplicación de consumo de Java KCL desde GitHub, visite la página del proyecto de [ejemplo de KCL para Java](https://github.com/aws/aws-sdk-java/tree/master/src/samples/AmazonKinesis) en. GitHub La aplicación de muestra utiliza [Apache Commons Logging](http://commons.apache.org/proper/commons-logging/guide.html). Puede cambiar la configuración de registro en el método estático `configure` definido en el archivo `AmazonKinesisApplicationSample.java`. *Para obtener más información sobre cómo utilizar el registro de Apache Commons con Log4j y aplicaciones AWS Java, consulte [Registrar con Log4j](https://docs.aws.amazon.com/sdk-for-java/latest/developer-guide/java-dg-logging.html) en la guía para desarrolladores.AWS SDK para Java * Debe completar las siguientes tareas a la hora de implementar una aplicación de consumo de KCL en Java: **Topics** + [ ## Implemente los métodos del procesador IRecord ](#kinesis-record-processor-implementation-interface-java) + [ ## Implemente una fábrica de clases para la interfaz IRecord del procesador ](#kinesis-record-processor-implementation-factory-java) + [ ## Crear un proceso de trabajo ](#kcl-java-worker) + [ ## Modificar las propiedades de configuración ](#kinesis-record-processor-initialization-java) + [ ## Migrar a la versión 2 de la interfaz del procesador de registros ](#kcl-java-v2-migration) ## Implemente los métodos del procesador IRecord KCL es compatible actualmente con dos versiones de la interfaz de `IRecordProcessor`: la interfaz original está disponible con la primera versión de KCL y la versión 2 está disponible a partir de la versión 1.5.0 de KCL. Ambas interfaces son totalmente compatibles. La elección dependerá de su situación específica. Consulte sus javadocs locales o el código fuente para ver todas las diferencias. En las siguientes secciones se describe la implementación mínima introductoria. **Topics** + [ ### Interfaz original (versión 1) ](#kcl-java-interface-original) + [ ### Interfaz actualizada (versión 2) ](#kcl-java-interface-v2) ### Interfaz original (versión 1) La interfaz original `IRecordProcessor` (`package com.amazonaws.services.kinesis.clientlibrary.interfaces`) expone los siguientes métodos del procesador de registros que el consumidor debe implementar. En la muestra se presentan implementaciones que puede utilizar como punto de partida (consulte `AmazonKinesisApplicationSampleRecordProcessor.java`). ``` public void initialize(String shardId) public void processRecords(List records, IRecordProcessorCheckpointer checkpointer) public void shutdown(IRecordProcessorCheckpointer checkpointer, ShutdownReason reason) ``` **inicializar** KCL llama al método `initialize` cuando se crea una instancia del procesador de registros y pasa un ID de partición específico como parámetro. Este procesador de registros procesa solo este fragmento, y normalmente también se produce la situación contraria (este fragmento solo es procesado por este procesador de registros). Sin embargo, el consumidor debe contar con la posibilidad de que un registro de datos pueda ser procesado más de una vez. Kinesis Data Streams tiene una semántica de *al menos una vez*, lo que significa que cada registro de datos de una partición se procesa al menos una vez por parte de un proceso de trabajo del consumidor. Para obtener más información sobre los casos en los que un fragmento en particular puede ser procesado por más de un proceso de trabajo, consulte [Utilizar la nueva partición, el escalado y el procesamiento paralelo para cambiar el número de particiones](kinesis-record-processor-scaling.md). ``` public void initialize(String shardId) ``` **processRecords** KCL llama al método `processRecords` y pasa una lista de registros de datos desde la partición especificada por el método `initialize(shardId)`. El procesador de registros procesa los datos en estos registros según la semántica del consumidor. Por ejemplo, el proceso de trabajo podría realizar una transformación de los datos y, a continuación, almacenar el resultado en un bucket de Amazon Simple Storage Service (Amazon S3). ``` public void processRecords(List records, IRecordProcessorCheckpointer checkpointer) ``` Además de los datos en sí, el registro contiene un número secuencial y una clave de partición. El proceso de trabajo puede utilizar estos valores al procesar los datos. Por ejemplo, el proceso de trabajo podría elegir el bucket de S3 en el que almacenar los datos en función del valor de la clave de partición. La clase `Record` expone los siguientes métodos que proporcionan acceso a los datos, el número secuencial y la clave de partición del registro. ``` record.getData() record.getSequenceNumber() record.getPartitionKey() ``` En el ejemplo, el método privado `processRecordsWithRetries` tiene un código que muestra cómo un proceso de trabajo puede obtener acceso a los datos, el número secuencial y la clave de partición del registro. Kinesis Data Streams requiere que el procesador de registros realice un seguimiento de los registros que ya se han procesado en una partición. KCL se ocupa de este seguimiento pasando un generador de puntos de verificación (`IRecordProcessorCheckpointer`) a `processRecords`. El procesador de registros llama al método `checkpoint` en esta interfaz para informar a KCL de su avance en el procesamiento de los registros de la partición. Si se produce un error en el proceso de trabajo, KCL utiliza esta información para reiniciar el procesamiento de la partición en el último registro procesado conocido. En el caso de una operación de división o fusión, KCL no comenzará a procesar las particiones nuevas hasta que los procesadores de las particiones originales hayan llamado a `checkpoint` para indicar que se ha completado el procesamiento en las particiones originales. Si no se pasa un parámetro, KCL supone que la llamada a `checkpoint` significa que todos los registros se han procesado, hasta el último registro pasado al procesador de registros. Por tanto, el procesador de registros solo debe llamar a `checkpoint` después de haber procesado todos los registros de la lista que se le ha pasado. Los procesadores de registros no necesitan llamar a `checkpoint` en cada llamada a `processRecords`. Un procesador podría, por ejemplo, llamar a `checkpoint` cada tercera vez que llame a `processRecords`. Puede especificar opcionalmente el número secuencial exacto de un registro como un parámetro para `checkpoint`. En este caso, KCL supone que todos los registros se han procesado exclusivamente hasta ese registro. En el ejemplo, el método privado `checkpoint` muestra cómo llamar a `IRecordProcessorCheckpointer.checkpoint` mediante la administración de excepciones y la lógica de reintentos apropiadas. KCL depende de `processRecords` para administrar cualquier excepción que surja del procesamiento de los registros de datos. Si `processRecords` genera una excepción, KCL omite los registros de datos que se pasaron antes de la excepción. Es decir, estos registros no se reenviarán al procesador de registros que generó la excepción ni a ningún otro procesador de registros en el consumidor. **shutdown** KCL llama al método `shutdown` cuando finaliza el procesamiento (el motivo del cierre es `TERMINATE`) o cuando el proceso de trabajo ya no responde (el motivo del cierre es `ZOMBIE`). ``` public void shutdown(IRecordProcessorCheckpointer checkpointer, ShutdownReason reason) ``` El procesamiento finaliza cuando el procesador de registros no recibe más registros desde el fragmento, ya sea porque el fragmento se ha dividido o fusionado o porque la secuencia se ha eliminado. KCL también pasa una interfaz `IRecordProcessorCheckpointer` a `shutdown`. Si el motivo del shutdown es `TERMINATE`, el procesador de registros debería terminar de procesar los registros de datos y llamar al método `checkpoint` en esta interfaz. ### Interfaz actualizada (versión 2) La interfaz actualizada `IRecordProcessor` (`package com.amazonaws.services.kinesis.clientlibrary.interfaces.v2`) expone los siguientes métodos del procesador de registros que el consumidor debe implementar: ``` void initialize(InitializationInput initializationInput) void processRecords(ProcessRecordsInput processRecordsInput) void shutdown(ShutdownInput shutdownInput) ``` Se puede obtener acceso a todos los argumentos de la versión original de la interfaz mediante métodos "get" en los objetos del contenedor. Por ejemplo, para recuperar la lista de registros en `processRecords()`, puede utilizar `processRecordsInput.getRecords()`. A partir de la versión 2 de esta interfaz (KCL 1.5.0 y posteriores), están disponibles las siguientes entradas nuevas, además de las entradas proporcionadas por la interfaz original: starting sequence number En el objeto `InitializationInput` que se pasa a la operación `initialize()` el número secuencial inicial a partir del cual se facilitan los registros a la instancia del procesador de registros. Este es el último número secuencial objeto de un punto de comprobación por parte de la instancia del procesador de registros que procesara anteriormente el mismo fragmento. Estos datos se ofrecen por si su aplicación necesitara esta información. pending checkpoint sequence number En el objeto `InitializationInput` que se pasa a la operación `initialize()`, el número secuencial pendiente de punto de comprobación (si hay alguno) que no se ha podido confirmar antes de que se detuviera la instancia anterior del procesador de registros. ## Implemente una fábrica de clases para la interfaz IRecord del procesador También necesitará implementar un generador para la clase que implementa los métodos del procesador de registros. Cuando el consumidor crea instancias del proceso de trabajo, pasa una referencia a este generador. La muestra implementa el generador de clases en el archivo `AmazonKinesisApplicationSampleRecordProcessorFactory.java` mediante la interfaz del procesador de registros original. Si desea que el generador de clases cree procesadores de registros de la versión 2, utilice el nombre de paquete `com.amazonaws.services.kinesis.clientlibrary.interfaces.v2`. ``` public class SampleRecordProcessorFactory implements IRecordProcessorFactory { /** * Constructor. */ public SampleRecordProcessorFactory() { super(); } /** * {@inheritDoc} */ @Override public IRecordProcessor createProcessor() { return new SampleRecordProcessor(); } } ``` ## Crear un proceso de trabajo Tal y como se ha explicado en [Implemente los métodos del procesador IRecord](#kinesis-record-processor-implementation-interface-java), hay dos versiones de la interfaz de procesador de registros de KCL para elegir, lo que afecta a la creación de un proceso de trabajo. La interfaz de procesador de registros original utiliza la siguiente estructura de código para crear un proceso de trabajo: ``` final KinesisClientLibConfiguration config = new KinesisClientLibConfiguration(...) final IRecordProcessorFactory recordProcessorFactory = new RecordProcessorFactory(); final Worker worker = new Worker(recordProcessorFactory, config); ``` Con la versión 2 del procesador de registros, puede utilizar la `Worker.Builder` para crear un proceso de trabajo sin preocuparse por qué constructor utilizar ni por el orden de los argumentos. La interfaz de procesador de registros actualizada utiliza la siguiente estructura de código para crear un proceso de trabajo: ``` final KinesisClientLibConfiguration config = new KinesisClientLibConfiguration(...) final IRecordProcessorFactory recordProcessorFactory = new RecordProcessorFactory(); final Worker worker = new Worker.Builder() .recordProcessorFactory(recordProcessorFactory) .config(config) .build(); ``` ## Modificar las propiedades de configuración En la muestra se proporcionan valores predeterminados para las propiedades de configuración. Este datos de configuración del proceso de trabajo se consolidan posteriormente en un objeto `KinesisClientLibConfiguration`. Este objeto y una referencia al generador de clases de `IRecordProcessor` se pasan en la llamada que crea la instancia del proceso de trabajo. Puede sobrescribir cualquiera de estas propiedades con sus propios valores a través de un archivo de propiedades de Java (consulte `AmazonKinesisApplicationSample.java`). ### Nombre de la aplicación KCL requiere un nombre de aplicación que sea único entre las aplicaciones y en las tablas de Amazon DynamoDB de la misma región. La biblioteca utiliza el valor del nombre de la aplicación de las siguientes formas: + Se entiende que los procesos de trabajo asociados a este nombre de aplicación operan de forma conjunta en la misma secuencia. Estos procesos de trabajo pueden distribuirse en varias instancias. Si ejecuta otra instancia del mismo código de aplicación, pero con otro nombre de aplicación, KCL considera que la segunda instancia es una aplicación completamente independiente de la otra que opera en el mismo flujo. + KCL crea una tabla de DynamoDB con el nombre de la aplicación y utiliza la tabla para actualizar la información de estado (como los puntos de verificación y el mapeo procesos de trabajo-particiones) para la aplicación. Cada aplicación tiene su propia tabla de DynamoDB. Para obtener más información, consulte [Usar una tabla de arrendamiento para realizar el seguimiento de las particiones procesadas por la aplicación de consumo de KCL](shared-throughput-kcl-consumers.md#shared-throughput-kcl-consumers-leasetable). ### Configuración de credenciales Debe poner sus AWS credenciales a disposición de uno de los proveedores de credenciales de la cadena de proveedores de credenciales predeterminada. Por ejemplo, si ejecuta su consumidor en una instancia de EC2, se recomienda que lance la instancia con un rol de IAM. Las credenciales de AWS que reflejan los permisos asociados a este rol de IAM se ponen a disposición de las aplicaciones de la instancia a través de los metadatos de esta. Esta es la forma más segura de administrar las credenciales para un consumidor que se ejecute en una instancia EC2. En primer lugar, la aplicación de muestra intenta recuperar las credenciales de IAM de los metadatos de la instancia: ``` credentialsProvider = new InstanceProfileCredentialsProvider(); ``` Si la aplicación de muestra no puede obtener credenciales de los metadatos de la instancia, intenta recuperar las credenciales desde un archivo de propiedades: ``` credentialsProvider = new ClasspathPropertiesFileCredentialsProvider(); ``` Para más información sobre los metadatos de instancia, consulte [Metadatos de instancia](https://docs.aws.amazon.com/AWSEC2/latest/UserGuide/ec2-instance-metadata.html) en la *Guía del usuario de Amazon EC2*. ### Uso de ID de proceso de trabajo para varias instancias El código de inicialización de muestra crea un ID para el proceso de trabajo, `workerId`, con el nombre del equipo local y un identificador global único anexo, tal y como se muestra en el siguiente fragmento de código. Este enfoque es compatible con un escenario con varias instancias de la aplicación consumidora ejecutándose en un único equipo. ``` String workerId = InetAddress.getLocalHost().getCanonicalHostName() + ":" + UUID.randomUUID(); ``` ## Migrar a la versión 2 de la interfaz del procesador de registros Si desea migrar código que utilice la interfaz original, además de los pasos descritos anteriormente, tendrá que seguir estos pasos: 1. Cambie la clase de su procesador de registros para importar la versión 2 de la interfaz del procesador de registros: ``` import com.amazonaws.services.kinesis.clientlibrary.interfaces.v2.IRecordProcessor; ``` 1. Cambiar las referencias a las entradas para usar métodos `get` en los objetos del contenedor. Por ejemplo, en la operación `shutdown()`, cambie "`checkpointer`" por "`shutdownInput.getCheckpointer()`". 1. Cambie la clase del generador de procesadores de registros para importar la interfaz del generador de procesadores de registros de la versión 2: ``` import com.amazonaws.services.kinesis.clientlibrary.interfaces.v2.IRecordProcessorFactory; ``` 1. Cambie la construcción del proceso de trabajo para usar `Worker.Builder`. Por ejemplo: ``` final Worker worker = new Worker.Builder() .recordProcessorFactory(recordProcessorFactory) .config(config) .build(); ``` # Desarrollar un consumidor de Kinesis Client Library en Node.js **importante** Las versiones 1.x y 2.x de la biblioteca de clientes de Amazon Kinesis (KCL) están obsoletas. KCL 1.x estará disponible el 30 de end-of-support enero de 2026. **Recomendamos** que migre las aplicaciones de KCL que utilizan la versión 1.x a la última versión de KCL antes del 30 de enero de 2026. Para encontrar la versión más reciente de KCL, consulte la página de la [biblioteca de clientes de Amazon Kinesis](https://github.com/awslabs/amazon-kinesis-client) en. GitHub Para obtener información sobre las versiones más recientes de KCL, consulte [Uso de Kinesis Client Library](kcl.md). Para obtener información sobre cómo migrar de KCL 1.x a KCL 3.x, consulte [Migración de KCL 1.x a KCL 3.x](kcl-migration-1-3.md). Puede utilizar Kinesis Client Library (KCL) para crear aplicaciones que procesen datos de los flujos de datos de Kinesis. Kinesis Client Library está disponible en varios idiomas. En este tema se habla de Node.js La KCL es una biblioteca de Java; el soporte para otros lenguajes además de Java se proporciona mediante una interfaz multilingüe llamada. *MultiLangDaemon* Este daemon está basado en Java y se ejecuta en segundo plano cuando se utiliza un lenguaje de KCL distinto de Java. Por lo tanto, si instala el KCL para Node.js y escribe su aplicación para consumidores completamente en Node.js, seguirá necesitando instalar Java en su sistema debido a la. MultiLangDaemon Además, MultiLangDaemon tiene algunos ajustes predeterminados que puede que tengas que personalizar para tu caso de uso, por ejemplo, la AWS región a la que se conecta. Para obtener más información sobre MultiLangDaemon esto GitHub, visita la página del [ MultiLangDaemon proyecto KCL](https://github.com/awslabs/amazon-kinesis-client/tree/v1.x/src/main/java/com/amazonaws/services/kinesis/multilang). Para descargar el archivo KCL de Node.js GitHub, vaya a la [biblioteca de clientes de Kinesis (Node.js)](https://github.com/awslabs/amazon-kinesis-client-nodejs). **Descargas de código de muestra** Hay dos códigos de muestra disponibles para la KCL en Node.js: + [basic-sample](https://github.com/awslabs/amazon-kinesis-client-nodejs/tree/master/samples/basic_sample) Se utiliza en las siguientes secciones para ilustrar los aspectos fundamentales de la creación de una aplicación de consumo de KCL en Node.js. + [click-stream-sample](https://github.com/awslabs/amazon-kinesis-client-nodejs/tree/master/samples/click_stream_sample) Es ligeramente más avanzado y utiliza una situación real. Para cuando se haya familiarizado con el código de muestra básico. Esta muestra no se trata aquí, pero tiene un archivo README con más información. Debe completar las siguientes tareas a la hora de implementar una aplicación de consumo de KCL en Node.js: **Topics** + [ ## Implementar el procesador de registros ](#kinesis-record-processor-implementation-interface-nodejs) + [ ## Modificar las propiedades de configuración ](#kinesis-record-processor-initialization-nodejs) ## Implementar el procesador de registros El consumidor más sencillo posible que utilice KCL para Node.js debe implementar una función `recordProcessor`, que a su vez contenga las funciones `initialize`, `processRecords` y `shutdown`. En la muestra se presenta una implementación que puede utilizar como punto de partida (consulte `sample_kcl_app.js`). ``` function recordProcessor() { // return an object that implements initialize, processRecords and shutdown functions.} ``` **inicializar** KCL llama a la función `initialize` cuando se inicia el procesador de registros. Este procesador de registros procesa solo la ID de fragmento que se haya pasado como `initializeInput.shardId`, y normalmente también se produce la situación contraria (este fragmento solo es procesado por este procesador de registros). Sin embargo, el consumidor debe contar con la posibilidad de que un registro de datos pueda ser procesado más de una vez. Esto se debe a que Kinesis Data Streams tiene una semántica de *al menos una vez*, lo que significa que cada registro de datos de una partición se procesa al menos una vez por parte de un proceso de trabajo del consumidor. Para obtener más información sobre los casos en los que un fragmento en particular puede ser procesado por varios procesos de trabajo, consulte [Utilizar la nueva partición, el escalado y el procesamiento paralelo para cambiar el número de particiones](kinesis-record-processor-scaling.md). ``` initialize: function(initializeInput, completeCallback) ``` **processRecords** KCL llama a esta función con una entrada que contiene una lista de registros de datos de la partición especificados para la función `initialize`. El procesador de registros que implemente procesa los datos en estos registros según la semántica del consumidor. Por ejemplo, el proceso de trabajo podría realizar una transformación de los datos y, a continuación, almacenar el resultado en un bucket de Amazon Simple Storage Service (Amazon S3). ``` processRecords: function(processRecordsInput, completeCallback) ``` Además de los datos en sí, el registro también contiene un número secuencial y una clave de partición, que el proceso de trabajo puede utilizar al procesar los datos. Por ejemplo, el proceso de trabajo podría elegir el bucket de S3 en el que almacenar los datos en función del valor de la clave de partición. El diccionario `record` expone los siguientes pares clave-valor para obtener acceso a los datos, el número secuencial y la clave de partición del registro: ``` record.data record.sequenceNumber record.partitionKey ``` Tenga en cuenta que los datos se codifican en Base64. En el ejemplo básico, la función `processRecords` tiene un código que muestra cómo un proceso de trabajo puede obtener acceso a los datos, el número secuencial y la clave de partición del registro. Kinesis Data Streams requiere que el procesador de registros realice un seguimiento de los registros que ya se han procesado en una partición. KCL se ocupa de este seguimiento con un objeto `checkpointer` que pasa como `processRecordsInput.checkpointer`. Su procesador de registros llama a la función `checkpointer.checkpoint` para informar a KCL de su avance en el procesamiento de los registros de la partición. En el caso de que se produzca un error en el proceso de trabajo, KCL utiliza esta información al reiniciar el procesamiento de la partición para continuar desde el último registro procesado conocido. En el caso de una operación de división o fusión, KCL no comenzará a procesar las particiones nuevas hasta que los procesadores de las particiones originales hayan llamado a `checkpoint` para indicar que se ha completado el procesamiento en las particiones originales. Si no pasa el número secuencial a la función `checkpoint`, KCL supone que la llamada a `checkpoint` significa que todos los registros se han procesado, hasta el último registro pasado al procesador de registros. Por tanto, el procesador de registros debería llamar a `checkpoint` **solo** después de haber procesado todos los registros de la lista que se le ha pasado. Los procesadores de registros no necesitan llamar a `checkpoint` en cada llamada a `processRecords`. Un procesador podría, por ejemplo, realizar una llamada `checkpoint` una de cada tres veces o algún evento externo a su procesador de registros, como un verification/validation servicio personalizado que haya implementado. Puede especificar opcionalmente el número secuencial exacto de un registro como un parámetro para `checkpoint`. En este caso, KCL supone que todos los registros se han procesado exclusivamente hasta ese registro. La aplicación de muestra básica muestra la llamada más sencilla posible a la función `checkpointer.checkpoint`. Puede agregar otra lógica de creación de puntos de comprobación que necesite para su consumidor en este punto en la función. **shutdown** KCL llama a la función `shutdown` cuando finaliza el procesamiento (`shutdownInput.reason` es `TERMINATE`) o cuando el proceso de trabajo ya no responde (`shutdownInput.reason` es `ZOMBIE`). ``` shutdown: function(shutdownInput, completeCallback) ``` El procesamiento finaliza cuando el procesador de registros no recibe más registros desde el fragmento, ya sea porque el fragmento se ha dividido o fusionado o porque la secuencia se ha eliminado. KCL también pasa un objeto `shutdownInput.checkpointer` a `shutdown`. Si el motivo del cierre es `TERMINATE`, debe asegurarse de que el procesador de registros haya terminado de procesar los registros de datos y después llame a la función `checkpoint` en esta interfaz. ## Modificar las propiedades de configuración En la muestra se proporcionan valores predeterminados para las propiedades de configuración. Puede sobrescribir cualquiera de estas propiedades con sus propios valores (consulte `sample.properties` en el ejemplo básico). ### Nombre de la aplicación KCL requiere una aplicación que sea única entre las aplicaciones y en las tablas de Amazon DynamoDB de la misma región. La biblioteca utiliza el valor del nombre de la aplicación de las siguientes formas: + Se entiende que los procesos de trabajo asociados a este nombre de aplicación operan de forma conjunta en la misma secuencia. Estos procesos de trabajo pueden distribuirse en varias instancias. Si ejecuta otra instancia del mismo código de aplicación, pero con otro nombre de aplicación, KCL considera que la segunda instancia es una aplicación completamente independiente de la otra que opera en el mismo flujo. + KCL crea una tabla de DynamoDB con el nombre de la aplicación y utiliza la tabla para actualizar la información de estado (como los puntos de verificación y el mapeo procesos de trabajo-particiones) para la aplicación. Cada aplicación tiene su propia tabla de DynamoDB. Para obtener más información, consulte [Usar una tabla de arrendamiento para realizar el seguimiento de las particiones procesadas por la aplicación de consumo de KCL](shared-throughput-kcl-consumers.md#shared-throughput-kcl-consumers-leasetable). ### Configuración de credenciales Debe poner sus AWS credenciales a disposición de uno de los proveedores de credenciales de la cadena de proveedores de credenciales predeterminada. Puede usar la propiedad `AWSCredentialsProvider` para configurar un proveedor de credenciales. El archivo `sample.properties` debe poner sus credenciales a disposición de uno de los proveedores de credenciales de la [cadena de proveedores de credenciales predeterminada](https://docs.aws.amazon.com/sdk-for-java/latest/reference/com/amazonaws/auth/DefaultAWSCredentialsProviderChain.html). Si ejecuta su consumidor en una instancia de Amazon EC2, le recomendamos que configure la instancia con un rol de IAM. AWS Las credenciales que reflejan los permisos asociados a esta función de IAM se ponen a disposición de las aplicaciones de la instancia a través de los metadatos de la instancia. Esta es la forma más segura de administrar las credenciales para una aplicación consumidora que se ejecute en una instancia EC2. En el siguiente ejemplo, se configura KCL para procesar un flujo de datos de Kinesis denominado `kclnodejssample` utilizando el procesador de registros facilitado en `sample_kcl_app.js`: ``` # The Node.js executable script executableName = node sample_kcl_app.js # The name of an Amazon Kinesis stream to process streamName = kclnodejssample # Unique KCL application name applicationName = kclnodejssample # Use default AWS credentials provider chain AWSCredentialsProvider = DefaultAWSCredentialsProviderChain # Read from the beginning of the stream initialPositionInStream = TRIM_HORIZON ``` # Desarrollar un consumidor de Kinesis Client Library en .NET **importante** Las versiones 1.x y 2.x de la biblioteca de clientes de Amazon Kinesis (KCL) están obsoletas. KCL 1.x estará disponible el 30 de end-of-support enero de 2026. **Recomendamos** que migre las aplicaciones de KCL que utilizan la versión 1.x a la última versión de KCL antes del 30 de enero de 2026. Para encontrar la versión más reciente de KCL, consulte la página de la [biblioteca de clientes de Amazon Kinesis](https://github.com/awslabs/amazon-kinesis-client) en. GitHub Para obtener información sobre las versiones más recientes de KCL, consulte [Uso de Kinesis Client Library](kcl.md). Para obtener información sobre cómo migrar de KCL 1.x a KCL 3.x, consulte [Migración de KCL 1.x a KCL 3.x](kcl-migration-1-3.md). Puede utilizar Kinesis Client Library (KCL) para crear aplicaciones que procesen datos de los flujos de datos de Kinesis. Kinesis Client Library está disponible en varios idiomas. En este tema se habla de .NET. La KCL es una biblioteca de Java; el soporte para otros lenguajes además de Java se proporciona mediante una interfaz multilingüe llamada. *MultiLangDaemon* Este daemon está basado en Java y se ejecuta en segundo plano cuando se utiliza un lenguaje de KCL distinto de Java. Por lo tanto, si instala la KCL para.NET y escribe la aplicación para el usuario en su totalidad en .NET, seguirá necesitando instalar Java en el sistema debido a la. MultiLangDaemon Además, MultiLangDaemon tiene algunos ajustes predeterminados que puede que tengas que personalizar para tu caso de uso, por ejemplo, la AWS región a la que se conecta. Para obtener más información sobre MultiLangDaemon esto GitHub, visita la página del [ MultiLangDaemon proyecto KCL](https://github.com/awslabs/amazon-kinesis-client/tree/v1.x/src/main/java/com/amazonaws/services/kinesis/multilang). Para descargar el KCL de.NET GitHub, vaya a la [biblioteca de clientes de Kinesis (](https://github.com/awslabs/amazon-kinesis-client-net).NET). Para descargar un código de muestra para una aplicación de consumo de KCL de.NET, visite la página de [ejemplos de proyectos de consumo de KCL para .NET](https://github.com/awslabs/amazon-kinesis-client-net/tree/master/SampleConsumer) en. GitHub Debe completar las siguientes tareas a la hora de implementar una aplicación de consumo de KCL en .NET: **Topics** + [ ## Implemente los métodos de la clase IRecord Processor ](#kinesis-record-processor-implementation-interface-dotnet) + [ ## Modificar las propiedades de configuración ](#kinesis-record-processor-initialization-dotnet) ## Implemente los métodos de la clase IRecord Processor El consumidor debe implementar los siguientes métodos para `IRecordProcessor`. En el consumidor de muestra se presentan implementaciones que puede utilizar como punto de partida (consulte la clase `SampleRecordProcessor` en `SampleConsumer/AmazonKinesisSampleConsumer.cs`). ``` public void Initialize(InitializationInput input) public void ProcessRecords(ProcessRecordsInput input) public void Shutdown(ShutdownInput input) ``` **Initialize** KCL llama a este método cuando se crea una instancia del procesador de registros y pasa un ID de partición específico en el parámetro `input` (`input.ShardId`). Este procesador de registros procesa solo este fragmento, y normalmente también se produce la situación contraria (este fragmento solo es procesado por este procesador de registros). Sin embargo, el consumidor debe contar con la posibilidad de que un registro de datos pueda ser procesado más de una vez. Esto se debe a que Kinesis Data Streams tiene una semántica de *al menos una vez*, lo que significa que cada registro de datos de una partición se procesa al menos una vez por parte de un proceso de trabajo del consumidor. Para obtener más información sobre los casos en los que un fragmento en particular puede ser procesado por varios procesos de trabajo, consulte [Utilizar la nueva partición, el escalado y el procesamiento paralelo para cambiar el número de particiones](kinesis-record-processor-scaling.md). ``` public void Initialize(InitializationInput input) ``` **ProcessRecords** KCL llama a este método y pasa una lista de registros de datos en el parámetro `input` (`input.Records`) desde la partición especificada por el método `Initialize`. El procesador de registros que implemente procesa los datos en estos registros según la semántica del consumidor. Por ejemplo, el proceso de trabajo podría realizar una transformación de los datos y, a continuación, almacenar el resultado en un bucket de Amazon Simple Storage Service (Amazon S3). ``` public void ProcessRecords(ProcessRecordsInput input) ``` Además de los datos en sí, el registro contiene un número secuencial y una clave de partición. El proceso de trabajo puede utilizar estos valores al procesar los datos. Por ejemplo, el proceso de trabajo podría elegir el bucket de S3 en el que almacenar los datos en función del valor de la clave de partición. La clase `Record` expone lo siguiente para obtener acceso a los datos, el número secuencial y la clave de partición del registro: ``` byte[] Record.Data string Record.SequenceNumber string Record.PartitionKey ``` En el ejemplo, el método `ProcessRecordsWithRetries` tiene un código que muestra cómo un proceso de trabajo puede obtener acceso a los datos, el número secuencial y la clave de partición del registro. Kinesis Data Streams requiere que el procesador de registros realice un seguimiento de los registros que ya se han procesado en una partición. KCL se ocupa de este seguimiento, pasando un objeto `Checkpointer` a `ProcessRecords` (`input.Checkpointer`). El procesador de registros llama al método `Checkpointer.Checkpoint` para informar a KCL de su avance en el procesamiento de los registros de la partición. Si se produce un error en el proceso de trabajo, KCL utiliza esta información para reiniciar el procesamiento de la partición en el último registro procesado conocido. En el caso de una operación de división o fusión, KCL no comenzará a procesar las particiones nuevas hasta que los procesadores de las particiones originales hayan llamado a `Checkpointer.Checkpoint` para indicar que se ha completado el procesamiento en las particiones originales. Si no se pasa un parámetro, KCL supone que la llamada a `Checkpointer.Checkpoint` significa que todos los registros se han procesado, hasta el último registro pasado al procesador de registros. Por tanto, el procesador de registros solo debe llamar a `Checkpointer.Checkpoint` después de haber procesado todos los registros de la lista que se le ha pasado. Los procesadores de registros no necesitan llamar a `Checkpointer.Checkpoint` en cada llamada a `ProcessRecords`. Un procesador podría, por ejemplo, llamar a `Checkpointer.Checkpoint` en cada tercera o cuarta llamada. Puede especificar opcionalmente el número secuencial exacto de un registro como un parámetro para `Checkpointer.Checkpoint`. En este caso, KCL supone que los registros se han procesado exclusivamente hasta ese registro. En el ejemplo, el método privado `Checkpoint(Checkpointer checkpointer)` muestra cómo llamar al método `Checkpointer.Checkpoint` mediante la administración de excepciones y la lógica de reintentos apropiadas. KCL para .NET administra las excepciones de forma diferente a las bibliotecas de KCL para el resto de lenguajes, ya que no administra las excepciones que surgen del procesamiento de los registros de datos. Las excepciones no detectadas procedentes del código del usuario harán que el programa se bloquee. **Apagado** KCL llama al método `Shutdown` cuando finaliza el procesamiento (el motivo del cierre es `TERMINATE`) o cuando el proceso de trabajo ya no responde (el valor de `input.Reason` del cierre es `ZOMBIE`). ``` public void Shutdown(ShutdownInput input) ``` El procesamiento finaliza cuando el procesador de registros no recibe más registros desde el fragmento, ya sea porque el fragmento se ha dividido o fusionado o porque la secuencia se ha eliminado. KCL también pasa un objeto `Checkpointer` a `shutdown`. Si el motivo del shutdown es `TERMINATE`, el procesador de registros debería terminar de procesar los registros de datos y llamar al método `checkpoint` en esta interfaz. ## Modificar las propiedades de configuración En el consumidor muestra se proporcionan valores predeterminados para las propiedades de configuración. Puede sobrescribir cualquiera de estas propiedades con sus propios valores (consulte `SampleConsumer/kcl.properties`). ### Nombre de la aplicación KCL requiere una aplicación que sea única entre las aplicaciones y en las tablas de Amazon DynamoDB de la misma región. La biblioteca utiliza el valor del nombre de la aplicación de las siguientes formas: + Se entiende que los procesos de trabajo asociados a este nombre de aplicación operan de forma conjunta en la misma secuencia. Estos procesos de trabajo pueden distribuirse en varias instancias. Si ejecuta otra instancia del mismo código de aplicación, pero con otro nombre de aplicación, KCL considera que la segunda instancia es una aplicación completamente independiente de la otra que opera en el mismo flujo. + KCL crea una tabla de DynamoDB con el nombre de la aplicación y utiliza la tabla para actualizar la información de estado (como los puntos de verificación y el mapeo procesos de trabajo-particiones) para la aplicación. Cada aplicación tiene su propia tabla de DynamoDB. Para obtener más información, consulte [Usar una tabla de arrendamiento para realizar el seguimiento de las particiones procesadas por la aplicación de consumo de KCL](shared-throughput-kcl-consumers.md#shared-throughput-kcl-consumers-leasetable). ### Configuración de credenciales Debe poner sus AWS credenciales a disposición de uno de los proveedores de credenciales de la cadena de proveedores de credenciales predeterminada. Puede usar la propiedad `AWSCredentialsProvider` para configurar un proveedor de credenciales. Las [propiedades de muestra](https://github.com/awslabs/amazon-kinesis-client-python/blob/master/samples/sample.properties) deben poner sus credenciales a disposición de uno de los proveedores de credenciales de la [cadena de proveedores de credenciales predeterminada](https://docs.aws.amazon.com/sdk-for-java/latest/reference/com/amazonaws/auth/DefaultAWSCredentialsProviderChain.html). Si ejecuta su aplicación de consumo en una instancia de EC2, se recomienda que configure la instancia con un rol de IAM. Las credenciales de AWS que reflejan los permisos asociados a este rol de IAM se ponen a disposición de las aplicaciones de la instancia a través de los metadatos de esta. Esta es la forma más segura de administrar las credenciales para un consumidor que se ejecute en una instancia EC2. El archivo de propiedades de ejemplo configura KCL para procesar un flujo de datos de Kinesis llamado “words” utilizando el procesador de registros facilitado en `AmazonKinesisSampleConsumer.cs`. # Desarrollar un consumidor de Kinesis Client Library en Python **importante** Las versiones 1.x y 2.x de la biblioteca de clientes de Amazon Kinesis (KCL) están obsoletas. KCL 1.x estará disponible el 30 de end-of-support enero de 2026. **Recomendamos** que migre las aplicaciones de KCL que utilizan la versión 1.x a la última versión de KCL antes del 30 de enero de 2026. Para encontrar la versión más reciente de KCL, consulte la página de la [biblioteca de clientes de Amazon Kinesis](https://github.com/awslabs/amazon-kinesis-client) en. GitHub Para obtener información sobre las versiones más recientes de KCL, consulte [Uso de Kinesis Client Library](kcl.md). Para obtener información sobre cómo migrar de KCL 1.x a KCL 3.x, consulte [Migración de KCL 1.x a KCL 3.x](kcl-migration-1-3.md). Puede utilizar Kinesis Client Library (KCL) para crear aplicaciones que procesen datos de los flujos de datos de Kinesis. Kinesis Client Library está disponible en varios idiomas. En este tema se habla de Python. La KCL es una biblioteca de Java; el soporte para otros lenguajes además de Java se proporciona mediante una interfaz multilingüe llamada. *MultiLangDaemon* Este daemon está basado en Java y se ejecuta en segundo plano cuando se utiliza un lenguaje de KCL distinto de Java. Por lo tanto, si instala el KCL para Python y escribe su aplicación de consumo completamente en Python, seguirá necesitando instalar Java en su sistema debido a la MultiLangDaemon. Además, MultiLangDaemon tiene algunos ajustes predeterminados que puede que tengas que personalizar para tu caso de uso, por ejemplo, la AWS región a la que se conecta. Para obtener más información sobre MultiLangDaemon esto GitHub, visita la página del [ MultiLangDaemon proyecto KCL](https://github.com/awslabs/amazon-kinesis-client/tree/v1.x/src/main/java/com/amazonaws/services/kinesis/multilang). Para descargar la KCL de Python GitHub, vaya a la biblioteca de [clientes de Kinesis (Python)](https://github.com/awslabs/amazon-kinesis-client-python). Para descargar un código de muestra para una aplicación de consumo de KCL para Python, vaya a la página del [proyecto de ejemplo de KCL para Python](https://github.com/awslabs/amazon-kinesis-client-python/tree/master/samples) en. GitHub Debe completar las siguientes tareas a la hora de implementar una aplicación de consumo de KCL en Python: **Topics** + [ ## Implementa los métodos de la clase RecordProcessor ](#kinesis-record-processor-implementation-interface-py) + [ ## Modificar las propiedades de configuración ](#kinesis-record-processor-initialization-py) ## Implementa los métodos de la clase RecordProcessor La clase `RecordProcess` debe ampliar la `RecordProcessorBase` para implementar los siguientes métodos. En la muestra se presentan implementaciones que puede utilizar como punto de partida (consulte `sample_kclpy_app.py`). ``` def initialize(self, shard_id) def process_records(self, records, checkpointer) def shutdown(self, checkpointer, reason) ``` **inicializar** KCL llama al método `initialize` cuando se crea una instancia del procesador de registros y pasa un ID de partición específico como parámetro. Este procesador de registros procesa solo este fragmento, y normalmente también se produce la situación contraria (este fragmento solo es procesado por este procesador de registros). Sin embargo, el consumidor debe contar con la posibilidad de que un registro de datos pueda ser procesado más de una vez. Esto se debe a que Kinesis Data Streams tiene una semántica de *al menos una vez*, lo que significa que cada registro de datos de una partición se procesa al menos una vez por parte de un proceso de trabajo del consumidor. Para obtener más información sobre los casos en los que un fragmento en particular puede ser procesado por más de un proceso de trabajo, consulte [Utilizar la nueva partición, el escalado y el procesamiento paralelo para cambiar el número de particiones](kinesis-record-processor-scaling.md). ``` def initialize(self, shard_id) ``` **process\$1records** KCL llama a este método y pasa una lista de registros de datos de la partición especificada por el método `initialize`. El procesador de registros que implemente procesa los datos en estos registros según la semántica del consumidor. Por ejemplo, el proceso de trabajo podría realizar una transformación de los datos y, a continuación, almacenar el resultado en un bucket de Amazon Simple Storage Service (Amazon S3). ``` def process_records(self, records, checkpointer) ``` Además de los datos en sí, el registro contiene un número secuencial y una clave de partición. El proceso de trabajo puede utilizar estos valores al procesar los datos. Por ejemplo, el proceso de trabajo podría elegir el bucket de S3 en el que almacenar los datos en función del valor de la clave de partición. El diccionario `record` expone los siguientes pares clave-valor para obtener acceso a los datos, el número secuencial y la clave de partición del registro: ``` record.get('data') record.get('sequenceNumber') record.get('partitionKey') ``` Tenga en cuenta que los datos se codifican en Base64. En el ejemplo, el método `process_records` tiene un código que muestra cómo un proceso de trabajo puede obtener acceso a los datos, el número secuencial y la clave de partición del registro. Kinesis Data Streams requiere que el procesador de registros realice un seguimiento de los registros que ya se han procesado en una partición. KCL se ocupa de este seguimiento, pasando un objeto `Checkpointer` a `process_records`. El procesador de registros llama al método `checkpoint` en este objeto para informar a KCL de su avance en el procesamiento de los registros de la partición. Si se produce un error en el proceso de trabajo, KCL utiliza esta información para reiniciar el procesamiento de la partición en el último registro procesado conocido. En el caso de una operación de división o fusión, KCL no comenzará a procesar las particiones nuevas hasta que los procesadores de las particiones originales hayan llamado a `checkpoint` para indicar que se ha completado el procesamiento en las particiones originales. Si no se pasa un parámetro, KCL supone que la llamada a `checkpoint` significa que todos los registros se han procesado, hasta el último registro pasado al procesador de registros. Por tanto, el procesador de registros solo debe llamar a `checkpoint` después de haber procesado todos los registros de la lista que se le ha pasado. Los procesadores de registros no necesitan llamar a `checkpoint` en cada llamada a `process_records`. Un procesador podría, por ejemplo, llamar a `checkpoint` cada tercera vez que llame. Puede especificar opcionalmente el número secuencial exacto de un registro como un parámetro para `checkpoint`. En este caso, KCL supone que todos los registros se han procesado exclusivamente hasta ese registro. En el ejemplo, el método privado `checkpoint` muestra cómo llamar al método `Checkpointer.checkpoint` mediante la administración de excepciones y la lógica de reintentos apropiadas. KCL depende de `process_records` para administrar cualquier excepción que surja del procesamiento de los registros de datos. Si `process_records` genera una excepción, KCL omite los registros de datos que se pasaron a `process_records` antes de la excepción. Es decir, estos registros no se reenviarán al procesador de registros que generó la excepción ni a ningún otro procesador de registros en el consumidor. **shutdown** KCL llama al método `shutdown` cuando finaliza el procesamiento (el motivo del cierre es `TERMINATE`) o cuando el proceso de trabajo ya no responde (el `reason` del cierre es `ZOMBIE`). ``` def shutdown(self, checkpointer, reason) ``` El procesamiento finaliza cuando el procesador de registros no recibe más registros desde el fragmento, ya sea porque el fragmento se ha dividido o fusionado o porque la secuencia se ha eliminado. KCL también pasa un objeto `Checkpointer` a `shutdown`. Si el `reason` del shutdown es `TERMINATE`, el procesador de registros debería terminar de procesar los registros de datos y llamar al método `checkpoint` en esta interfaz. ## Modificar las propiedades de configuración En la muestra se proporcionan valores predeterminados para las propiedades de configuración. Puede sobrescribir cualquiera de estas propiedades con sus propios valores (consulte `sample.properties`). ### Nombre de la aplicación KCL requiere un nombre de aplicación que sea único entre las aplicaciones y en las tablas de Amazon DynamoDB de la misma región. La biblioteca utiliza el valor del nombre de la aplicación de las siguientes formas: + Se supone que los procesos de trabajo que están asociados a este nombre de aplicación operan de forma conjunta en la misma secuencia. Estos procesos de trabajo pueden distribuirse en varias instancias. Si ejecuta otra instancia del mismo código de aplicación, pero con otro nombre de aplicación, KCL considera que la segunda instancia es una aplicación completamente independiente de la otra que opera en el mismo flujo. + KCL crea una tabla de DynamoDB con el nombre de la aplicación y utiliza la tabla para actualizar la información de estado (como los puntos de verificación y el mapeo procesos de trabajo-particiones) para la aplicación. Cada aplicación tiene su propia tabla de DynamoDB. Para obtener más información, consulte [Usar una tabla de arrendamiento para realizar el seguimiento de las particiones procesadas por la aplicación de consumo de KCL](shared-throughput-kcl-consumers.md#shared-throughput-kcl-consumers-leasetable). ### Configuración de credenciales Debe poner sus AWS credenciales a disposición de uno de los proveedores de credenciales de la cadena de proveedores de credenciales predeterminada. Puede usar la propiedad `AWSCredentialsProvider` para configurar un proveedor de credenciales. Las [propiedades de muestra](https://github.com/awslabs/amazon-kinesis-client-python/blob/master/samples/sample.properties) deben poner sus credenciales a disposición de uno de los proveedores de credenciales de la [cadena de proveedores de credenciales predeterminada](https://docs.aws.amazon.com/sdk-for-java/latest/reference/com/amazonaws/auth/DefaultAWSCredentialsProviderChain.html). Si ejecuta su aplicación de consumo en una instancia de Amazon EC2, se recomienda que configure la instancia con un rol de IAM. Las credenciales de AWS que reflejan los permisos asociados a este rol de IAM se ponen a disposición de las aplicaciones de la instancia a través de los metadatos de esta. Esta es la forma más segura de administrar las credenciales para una aplicación consumidora que se ejecute en una instancia EC2. El archivo de propiedades de ejemplo configura KCL para procesar un flujo de datos de Kinesis llamado “words” utilizando el procesador de registros facilitado en `sample_kclpy_app.py`. # Desarrollar un consumidor de Kinesis Client Library en Ruby **importante** Las versiones 1.x y 2.x de la biblioteca de clientes de Amazon Kinesis (KCL) están obsoletas. KCL 1.x estará disponible el 30 de end-of-support enero de 2026. **Recomendamos** que migre las aplicaciones de KCL que utilizan la versión 1.x a la última versión de KCL antes del 30 de enero de 2026. Para encontrar la versión más reciente de KCL, consulte la página de la [biblioteca de clientes de Amazon Kinesis](https://github.com/awslabs/amazon-kinesis-client) en. GitHub Para obtener información sobre las versiones más recientes de KCL, consulte [Uso de Kinesis Client Library](kcl.md). Para obtener información sobre cómo migrar de KCL 1.x a KCL 3.x, consulte [Migración de KCL 1.x a KCL 3.x](kcl-migration-1-3.md). Puede utilizar Kinesis Client Library (KCL) para crear aplicaciones que procesen datos de los flujos de datos de Kinesis. Kinesis Client Library está disponible en varios idiomas. En este tema se habla de Ruby. La KCL es una biblioteca de Java; el soporte para otros lenguajes además de Java se proporciona mediante una interfaz multilingüe llamada. *MultiLangDaemon* Este daemon está basado en Java y se ejecuta en segundo plano cuando se utiliza un lenguaje de KCL distinto de Java. Por lo tanto, si instala el KCL para Ruby y escribe su aplicación para consumidores completamente en Ruby, seguirá necesitando instalar Java en su sistema debido a la. MultiLangDaemon Además, MultiLangDaemon tiene algunos ajustes predeterminados que puede que tengas que personalizar para tu caso de uso, por ejemplo, la AWS región a la que se conecta. Para obtener más información sobre MultiLangDaemon esto GitHub, visita la página del [ MultiLangDaemon proyecto KCL](https://github.com/awslabs/amazon-kinesis-client/tree/v1.x/src/main/java/com/amazonaws/services/kinesis/multilang). Para descargar el KCL de Ruby GitHub, vaya a la [biblioteca de clientes de Kinesis (](https://github.com/awslabs/amazon-kinesis-client-ruby)Ruby). Para descargar un código de muestra para una aplicación de usuario de Ruby KCL, visite la página del proyecto de [ejemplo de KCL para Ruby](https://github.com/awslabs/amazon-kinesis-client-ruby/tree/master/samples) en. GitHub Para más información acerca de la biblioteca de soporte de Ruby en KCL, consulte la [documentación de Ruby Gems en KCL](http://www.rubydoc.info/gems/aws-kclrb). # Desarrollar consumidores de KCL 2.x **importante** Las versiones 1.x y 2.x de la biblioteca de clientes de Amazon Kinesis (KCL) están obsoletas. KCL 1.x estará disponible el 30 de end-of-support enero de 2026. **Recomendamos** que migre las aplicaciones de KCL que utilizan la versión 1.x a la última versión de KCL antes del 30 de enero de 2026. Para encontrar la versión más reciente de KCL, consulte la página de la [biblioteca de clientes de Amazon Kinesis](https://github.com/awslabs/amazon-kinesis-client) en. GitHub Para obtener información sobre las versiones más recientes de KCL, consulte [Uso de Kinesis Client Library](kcl.md). Para obtener información sobre cómo migrar de KCL 1.x a KCL 3.x, consulte [Migración de KCL 1.x a KCL 3.x](kcl-migration-1-3.md). En este tema se muestra cómo utilizar la versión 2.0 de Kinesis Client Library (KCL). Para más información acerca de KCL, consulte la información general que se proporciona en [Desarrollo de consumidores mediante Kinesis Client Library 1.x](https://docs.aws.amazon.com/streams/latest/dev/developing-consumers-with-kcl.html). Elija uno de los siguientes temas según la opción que desee utilizar. **Topics** + [ # Desarrollar un consumidor de Kinesis Client Library en Java ](kcl2-standard-consumer-java-example.md) + [ # Desarrollar un consumidor de Kinesis Client Library en Python ](kcl2-standard-consumer-python-example.md) + [ # Desarrollar consumidores de distribución mejorados con KCL 2.x ](building-enhanced-consumers-kcl-retired.md) # Desarrollar un consumidor de Kinesis Client Library en Java **importante** Las versiones 1.x y 2.x de la biblioteca de clientes de Amazon Kinesis (KCL) están obsoletas. La versión 1.x de KCL estará disponible el 30 de end-of-support enero de 2026. **Recomendamos** que migre las aplicaciones de KCL que utilizan la versión 1.x a la última versión de KCL antes del 30 de enero de 2026. Para encontrar la versión más reciente de KCL, consulte la página de la [biblioteca de clientes de Amazon Kinesis](https://github.com/awslabs/amazon-kinesis-client) en. GitHub Para obtener información sobre las versiones más recientes de KCL, consulte [Uso de Kinesis Client Library](kcl.md). Para obtener información sobre cómo migrar de KCL 1.x a KCL 3.x, consulte [Migración de KCL 1.x a KCL 3.x](kcl-migration-1-3.md). En el siguiente código, se muestra un ejemplo de implementación en Java de `ProcessorFactory` y `RecordProcessor`. Si desea aprovechar la característica de distribución ramificada mejorada, consulte [Uso de consumidores con la distribución ramificada mejorada](https://docs.aws.amazon.com/streams/latest/dev/building-enhanced-consumers-kcl-java.html). ``` /* * Copyright 2019 Amazon.com, Inc. or its affiliates. All Rights Reserved. * * Licensed under the Amazon Software License (the "License"). * You may not use this file except in compliance with the License. * A copy of the License is located at * * http://aws.amazon.com/asl/ * * or in the "license" file accompanying this file. This file is distributed * on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either * express or implied. See the License for the specific language governing * permissions and limitations under the License. */ /* * Copyright 2019 Amazon.com, Inc. or its affiliates. All Rights Reserved. * * Licensed under the Apache License, Version 2.0 (the "License"). * You may not use this file except in compliance with the License. * A copy of the License is located at * * http://www.apache.org/licenses/LICENSE-2.0 * * or in the "license" file accompanying this file. This file is distributed * on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either * express or implied. See the License for the specific language governing * permissions and limitations under the License. */ import java.io.BufferedReader; import java.io.IOException; import java.io.InputStreamReader; import java.util.UUID; import java.util.concurrent.ExecutionException; import java.util.concurrent.Executors; import java.util.concurrent.Future; import java.util.concurrent.ScheduledExecutorService; import java.util.concurrent.ScheduledFuture; import java.util.concurrent.TimeUnit; import java.util.concurrent.TimeoutException; import org.apache.commons.lang3.ObjectUtils; import org.apache.commons.lang3.RandomStringUtils; import org.apache.commons.lang3.RandomUtils; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.slf4j.MDC; import software.amazon.awssdk.core.SdkBytes; import software.amazon.awssdk.regions.Region; import software.amazon.awssdk.services.cloudwatch.CloudWatchAsyncClient; import software.amazon.awssdk.services.dynamodb.DynamoDbAsyncClient; import software.amazon.awssdk.services.kinesis.KinesisAsyncClient; import software.amazon.awssdk.services.kinesis.model.PutRecordRequest; import software.amazon.kinesis.common.ConfigsBuilder; import software.amazon.kinesis.common.KinesisClientUtil; import software.amazon.kinesis.coordinator.Scheduler; import software.amazon.kinesis.exceptions.InvalidStateException; import software.amazon.kinesis.exceptions.ShutdownException; import software.amazon.kinesis.lifecycle.events.InitializationInput; import software.amazon.kinesis.lifecycle.events.LeaseLostInput; import software.amazon.kinesis.lifecycle.events.ProcessRecordsInput; import software.amazon.kinesis.lifecycle.events.ShardEndedInput; import software.amazon.kinesis.lifecycle.events.ShutdownRequestedInput; import software.amazon.kinesis.processor.ShardRecordProcessor; import software.amazon.kinesis.processor.ShardRecordProcessorFactory; import software.amazon.kinesis.retrieval.polling.PollingConfig; /** * This class will run a simple app that uses the KCL to read data and uses the AWS SDK to publish data. * Before running this program you must first create a Kinesis stream through the AWS console or AWS SDK. */ public class SampleSingle { private static final Logger log = LoggerFactory.getLogger(SampleSingle.class); /** * Invoke the main method with 2 args: the stream name and (optionally) the region. * Verifies valid inputs and then starts running the app. */ public static void main(String... args) { if (args.length < 1) { log.error("At a minimum, the stream name is required as the first argument. The Region may be specified as the second argument."); System.exit(1); } String streamName = args[0]; String region = null; if (args.length > 1) { region = args[1]; } new SampleSingle(streamName, region).run(); } private final String streamName; private final Region region; private final KinesisAsyncClient kinesisClient; /** * Constructor sets streamName and region. It also creates a KinesisClient object to send data to Kinesis. * This KinesisClient is used to send dummy data so that the consumer has something to read; it is also used * indirectly by the KCL to handle the consumption of the data. */ private SampleSingle(String streamName, String region) { this.streamName = streamName; this.region = Region.of(ObjectUtils.firstNonNull(region, "us-east-2")); this.kinesisClient = KinesisClientUtil.createKinesisAsyncClient(KinesisAsyncClient.builder().region(this.region)); } private void run() { /** * Sends dummy data to Kinesis. Not relevant to consuming the data with the KCL */ ScheduledExecutorService producerExecutor = Executors.newSingleThreadScheduledExecutor(); ScheduledFuture producerFuture = producerExecutor.scheduleAtFixedRate(this::publishRecord, 10, 1, TimeUnit.SECONDS); /** * Sets up configuration for the KCL, including DynamoDB and CloudWatch dependencies. The final argument, a * ShardRecordProcessorFactory, is where the logic for record processing lives, and is located in a private * class below. */ DynamoDbAsyncClient dynamoClient = DynamoDbAsyncClient.builder().region(region).build(); CloudWatchAsyncClient cloudWatchClient = CloudWatchAsyncClient.builder().region(region).build(); ConfigsBuilder configsBuilder = new ConfigsBuilder(streamName, streamName, kinesisClient, dynamoClient, cloudWatchClient, UUID.randomUUID().toString(), new SampleRecordProcessorFactory()); /** * The Scheduler (also called Worker in earlier versions of the KCL) is the entry point to the KCL. This * instance is configured with defaults provided by the ConfigsBuilder. */ Scheduler scheduler = new Scheduler( configsBuilder.checkpointConfig(), configsBuilder.coordinatorConfig(), configsBuilder.leaseManagementConfig(), configsBuilder.lifecycleConfig(), configsBuilder.metricsConfig(), configsBuilder.processorConfig(), configsBuilder.retrievalConfig().retrievalSpecificConfig(new PollingConfig(streamName, kinesisClient)) ); /** * Kickoff the Scheduler. Record processing of the stream of dummy data will continue indefinitely * until an exit is triggered. */ Thread schedulerThread = new Thread(scheduler); schedulerThread.setDaemon(true); schedulerThread.start(); /** * Allows termination of app by pressing Enter. */ System.out.println("Press enter to shutdown"); BufferedReader reader = new BufferedReader(new InputStreamReader(System.in)); try { reader.readLine(); } catch (IOException ioex) { log.error("Caught exception while waiting for confirm. Shutting down.", ioex); } /** * Stops sending dummy data. */ log.info("Cancelling producer and shutting down executor."); producerFuture.cancel(true); producerExecutor.shutdownNow(); /** * Stops consuming data. Finishes processing the current batch of data already received from Kinesis * before shutting down. */ Future gracefulShutdownFuture = scheduler.startGracefulShutdown(); log.info("Waiting up to 20 seconds for shutdown to complete."); try { gracefulShutdownFuture.get(20, TimeUnit.SECONDS); } catch (InterruptedException e) { log.info("Interrupted while waiting for graceful shutdown. Continuing."); } catch (ExecutionException e) { log.error("Exception while executing graceful shutdown.", e); } catch (TimeoutException e) { log.error("Timeout while waiting for shutdown. Scheduler may not have exited."); } log.info("Completed, shutting down now."); } /** * Sends a single record of dummy data to Kinesis. */ private void publishRecord() { PutRecordRequest request = PutRecordRequest.builder() .partitionKey(RandomStringUtils.randomAlphabetic(5, 20)) .streamName(streamName) .data(SdkBytes.fromByteArray(RandomUtils.nextBytes(10))) .build(); try { kinesisClient.putRecord(request).get(); } catch (InterruptedException e) { log.info("Interrupted, assuming shutdown."); } catch (ExecutionException e) { log.error("Exception while sending data to Kinesis. Will try again next cycle.", e); } } private static class SampleRecordProcessorFactory implements ShardRecordProcessorFactory { public ShardRecordProcessor shardRecordProcessor() { return new SampleRecordProcessor(); } } /** * The implementation of the ShardRecordProcessor interface is where the heart of the record processing logic lives. * In this example all we do to 'process' is log info about the records. */ private static class SampleRecordProcessor implements ShardRecordProcessor { private static final String SHARD_ID_MDC_KEY = "ShardId"; private static final Logger log = LoggerFactory.getLogger(SampleRecordProcessor.class); private String shardId; /** * Invoked by the KCL before data records are delivered to the ShardRecordProcessor instance (via * processRecords). In this example we do nothing except some logging. * * @param initializationInput Provides information related to initialization. */ public void initialize(InitializationInput initializationInput) { shardId = initializationInput.shardId(); MDC.put(SHARD_ID_MDC_KEY, shardId); try { log.info("Initializing @ Sequence: {}", initializationInput.extendedSequenceNumber()); } finally { MDC.remove(SHARD_ID_MDC_KEY); } } /** * Handles record processing logic. The Amazon Kinesis Client Library will invoke this method to deliver * data records to the application. In this example we simply log our records. * * @param processRecordsInput Provides the records to be processed as well as information and capabilities * related to them (e.g. checkpointing). */ public void processRecords(ProcessRecordsInput processRecordsInput) { MDC.put(SHARD_ID_MDC_KEY, shardId); try { log.info("Processing {} record(s)", processRecordsInput.records().size()); processRecordsInput.records().forEach(r -> log.info("Processing record pk: {} -- Seq: {}", r.partitionKey(), r.sequenceNumber())); } catch (Throwable t) { log.error("Caught throwable while processing records. Aborting."); Runtime.getRuntime().halt(1); } finally { MDC.remove(SHARD_ID_MDC_KEY); } } /** Called when the lease tied to this record processor has been lost. Once the lease has been lost, * the record processor can no longer checkpoint. * * @param leaseLostInput Provides access to functions and data related to the loss of the lease. */ public void leaseLost(LeaseLostInput leaseLostInput) { MDC.put(SHARD_ID_MDC_KEY, shardId); try { log.info("Lost lease, so terminating."); } finally { MDC.remove(SHARD_ID_MDC_KEY); } } /** * Called when all data on this shard has been processed. Checkpointing must occur in the method for record * processing to be considered complete; an exception will be thrown otherwise. * * @param shardEndedInput Provides access to a checkpointer method for completing processing of the shard. */ public void shardEnded(ShardEndedInput shardEndedInput) { MDC.put(SHARD_ID_MDC_KEY, shardId); try { log.info("Reached shard end checkpointing."); shardEndedInput.checkpointer().checkpoint(); } catch (ShutdownException | InvalidStateException e) { log.error("Exception while checkpointing at shard end. Giving up.", e); } finally { MDC.remove(SHARD_ID_MDC_KEY); } } /** * Invoked when Scheduler has been requested to shut down (i.e. we decide to stop running the app by pressing * Enter). Checkpoints and logs the data a final time. * * @param shutdownRequestedInput Provides access to a checkpointer, allowing a record processor to checkpoint * before the shutdown is completed. */ public void shutdownRequested(ShutdownRequestedInput shutdownRequestedInput) { MDC.put(SHARD_ID_MDC_KEY, shardId); try { log.info("Scheduler is shutting down, checkpointing."); shutdownRequestedInput.checkpointer().checkpoint(); } catch (ShutdownException | InvalidStateException e) { log.error("Exception while checkpointing at requested shutdown. Giving up.", e); } finally { MDC.remove(SHARD_ID_MDC_KEY); } } } } ``` # Desarrollar un consumidor de Kinesis Client Library en Python **importante** Las versiones 1.x y 2.x de la biblioteca de clientes de Amazon Kinesis (KCL) están obsoletas. La versión 1.x de KCL estará disponible el 30 de end-of-support enero de 2026. **Recomendamos** que migre las aplicaciones de KCL que utilizan la versión 1.x a la última versión de KCL antes del 30 de enero de 2026. Para encontrar la versión más reciente de KCL, consulte la página de la [biblioteca de clientes de Amazon Kinesis](https://github.com/awslabs/amazon-kinesis-client) en. GitHub Para obtener información sobre las versiones más recientes de KCL, consulte [Uso de Kinesis Client Library](kcl.md). Para obtener información sobre cómo migrar de KCL 1.x a KCL 3.x, consulte [Migración de KCL 1.x a KCL 3.x](kcl-migration-1-3.md). Puede utilizar Kinesis Client Library (KCL) para crear aplicaciones que procesen datos de los flujos de datos de Kinesis. Kinesis Client Library está disponible en varios idiomas. En este tema se habla de Python. La KCL es una biblioteca de Java; el soporte para otros lenguajes además de Java se proporciona mediante una interfaz multilingüe llamada. *MultiLangDaemon* Este daemon está basado en Java y se ejecuta en segundo plano cuando se utiliza un lenguaje de KCL distinto de Java. Por lo tanto, si instala el KCL para Python y escribe su aplicación de consumo completamente en Python, seguirá necesitando instalar Java en su sistema debido a la MultiLangDaemon. Además, MultiLangDaemon tiene algunos ajustes predeterminados que puede que tengas que personalizar para tu caso de uso, por ejemplo, la AWS región a la que se conecta. Para obtener más información sobre MultiLangDaemon esto GitHub, visita la página del [ MultiLangDaemon proyecto KCL](https://github.com/awslabs/amazon-kinesis-client/tree/v1.x/src/main/java/com/amazonaws/services/kinesis/multilang). Para descargar la KCL de Python GitHub, vaya a la biblioteca de [clientes de Kinesis (Python)](https://github.com/awslabs/amazon-kinesis-client-python). Para descargar un código de muestra para una aplicación de consumo de KCL para Python, vaya a la página del [proyecto de ejemplo de KCL para Python](https://github.com/awslabs/amazon-kinesis-client-python/tree/master/samples) en. GitHub Debe completar las siguientes tareas a la hora de implementar una aplicación de consumo de KCL en Python: **Topics** + [ ## Implementa los métodos de la clase RecordProcessor ](#kinesis-record-processor-implementation-interface-py) + [ ## Modificar las propiedades de configuración ](#kinesis-record-processor-initialization-py) ## Implementa los métodos de la clase RecordProcessor La clase `RecordProcess` debe ampliar la `RecordProcessorBase` para implementar los siguientes métodos: ``` initialize process_records shutdown_requested ``` Este ejemplo proporciona implementaciones que puede utilizar como punto de partida. ``` #!/usr/bin/env python # Copyright 2014-2015 Amazon.com, Inc. or its affiliates. All Rights Reserved. # # Licensed under the Amazon Software License (the "License"). # You may not use this file except in compliance with the License. # A copy of the License is located at # # http://aws.amazon.com/asl/ # # or in the "license" file accompanying this file. This file is distributed # on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either # express or implied. See the License for the specific language governing # permissions and limitations under the License. from __future__ import print_function import sys import time from amazon_kclpy import kcl from amazon_kclpy.v3 import processor class RecordProcessor(processor.RecordProcessorBase): """ A RecordProcessor processes data from a shard in a stream. Its methods will be called with this pattern: * initialize will be called once * process_records will be called zero or more times * shutdown will be called if this MultiLangDaemon instance loses the lease to this shard, or the shard ends due a scaling change. """ def __init__(self): self._SLEEP_SECONDS = 5 self._CHECKPOINT_RETRIES = 5 self._CHECKPOINT_FREQ_SECONDS = 60 self._largest_seq = (None, None) self._largest_sub_seq = None self._last_checkpoint_time = None def log(self, message): sys.stderr.write(message) def initialize(self, initialize_input): """ Called once by a KCLProcess before any calls to process_records :param amazon_kclpy.messages.InitializeInput initialize_input: Information about the lease that this record processor has been assigned. """ self._largest_seq = (None, None) self._last_checkpoint_time = time.time() def checkpoint(self, checkpointer, sequence_number=None, sub_sequence_number=None): """ Checkpoints with retries on retryable exceptions. :param amazon_kclpy.kcl.Checkpointer checkpointer: the checkpointer provided to either process_records or shutdown :param str or None sequence_number: the sequence number to checkpoint at. :param int or None sub_sequence_number: the sub sequence number to checkpoint at. """ for n in range(0, self._CHECKPOINT_RETRIES): try: checkpointer.checkpoint(sequence_number, sub_sequence_number) return except kcl.CheckpointError as e: if 'ShutdownException' == e.value: # # A ShutdownException indicates that this record processor should be shutdown. This is due to # some failover event, e.g. another MultiLangDaemon has taken the lease for this shard. # print('Encountered shutdown exception, skipping checkpoint') return elif 'ThrottlingException' == e.value: # # A ThrottlingException indicates that one of our dependencies is is over burdened, e.g. too many # dynamo writes. We will sleep temporarily to let it recover. # if self._CHECKPOINT_RETRIES - 1 == n: sys.stderr.write('Failed to checkpoint after {n} attempts, giving up.\n'.format(n=n)) return else: print('Was throttled while checkpointing, will attempt again in {s} seconds' .format(s=self._SLEEP_SECONDS)) elif 'InvalidStateException' == e.value: sys.stderr.write('MultiLangDaemon reported an invalid state while checkpointing.\n') else: # Some other error sys.stderr.write('Encountered an error while checkpointing, error was {e}.\n'.format(e=e)) time.sleep(self._SLEEP_SECONDS) def process_record(self, data, partition_key, sequence_number, sub_sequence_number): """ Called for each record that is passed to process_records. :param str data: The blob of data that was contained in the record. :param str partition_key: The key associated with this recod. :param int sequence_number: The sequence number associated with this record. :param int sub_sequence_number: the sub sequence number associated with this record. """ #################################### # Insert your processing logic here #################################### self.log("Record (Partition Key: {pk}, Sequence Number: {seq}, Subsequence Number: {sseq}, Data Size: {ds}" .format(pk=partition_key, seq=sequence_number, sseq=sub_sequence_number, ds=len(data))) def should_update_sequence(self, sequence_number, sub_sequence_number): """ Determines whether a new larger sequence number is available :param int sequence_number: the sequence number from the current record :param int sub_sequence_number: the sub sequence number from the current record :return boolean: true if the largest sequence should be updated, false otherwise """ return self._largest_seq == (None, None) or sequence_number > self._largest_seq[0] or \ (sequence_number == self._largest_seq[0] and sub_sequence_number > self._largest_seq[1]) def process_records(self, process_records_input): """ Called by a KCLProcess with a list of records to be processed and a checkpointer which accepts sequence numbers from the records to indicate where in the stream to checkpoint. :param amazon_kclpy.messages.ProcessRecordsInput process_records_input: the records, and metadata about the records. """ try: for record in process_records_input.records: data = record.binary_data seq = int(record.sequence_number) sub_seq = record.sub_sequence_number key = record.partition_key self.process_record(data, key, seq, sub_seq) if self.should_update_sequence(seq, sub_seq): self._largest_seq = (seq, sub_seq) # # Checkpoints every self._CHECKPOINT_FREQ_SECONDS seconds # if time.time() - self._last_checkpoint_time > self._CHECKPOINT_FREQ_SECONDS: self.checkpoint(process_records_input.checkpointer, str(self._largest_seq[0]), self._largest_seq[1]) self._last_checkpoint_time = time.time() except Exception as e: self.log("Encountered an exception while processing records. Exception was {e}\n".format(e=e)) def lease_lost(self, lease_lost_input): self.log("Lease has been lost") def shard_ended(self, shard_ended_input): self.log("Shard has ended checkpointing") shard_ended_input.checkpointer.checkpoint() def shutdown_requested(self, shutdown_requested_input): self.log("Shutdown has been requested, checkpointing.") shutdown_requested_input.checkpointer.checkpoint() if __name__ == "__main__": kcl_process = kcl.KCLProcess(RecordProcessor()) kcl_process.run() ``` ## Modificar las propiedades de configuración En el ejemplo se proporcionan valores predeterminados para las propiedades de configuración, como los que se muestran en el siguiente script. Puede sobrescribir cualquiera de estas propiedades con sus propios valores. ``` # The script that abides by the multi-language protocol. This script will # be executed by the MultiLangDaemon, which will communicate with this script # over STDIN and STDOUT according to the multi-language protocol. executableName = sample_kclpy_app.py # The name of an Amazon Kinesis stream to process. streamName = words # Used by the KCL as the name of this application. Will be used as the name # of an Amazon DynamoDB table which will store the lease and checkpoint # information for workers with this application name applicationName = PythonKCLSample # Users can change the credentials provider the KCL will use to retrieve credentials. # The DefaultAWSCredentialsProviderChain checks several other providers, which is # described here: # http://docs.aws.amazon.com/AWSJavaSDK/latest/javadoc/com/amazonaws/auth/DefaultAWSCredentialsProviderChain.html AWSCredentialsProvider = DefaultAWSCredentialsProviderChain # Appended to the user agent of the KCL. Does not impact the functionality of the # KCL in any other way. processingLanguage = python/2.7 # Valid options at TRIM_HORIZON or LATEST. # See http://docs.aws.amazon.com/kinesis/latest/APIReference/API_GetShardIterator.html#API_GetShardIterator_RequestSyntax initialPositionInStream = TRIM_HORIZON # The following properties are also available for configuring the KCL Worker that is created # by the MultiLangDaemon. # The KCL defaults to us-east-1 #regionName = us-east-1 # Fail over time in milliseconds. A worker which does not renew it's lease within this time interval # will be regarded as having problems and it's shards will be assigned to other workers. # For applications that have a large number of shards, this msy be set to a higher number to reduce # the number of DynamoDB IOPS required for tracking leases #failoverTimeMillis = 10000 # A worker id that uniquely identifies this worker among all workers using the same applicationName # If this isn't provided a MultiLangDaemon instance will assign a unique workerId to itself. #workerId = # Shard sync interval in milliseconds - e.g. wait for this long between shard sync tasks. #shardSyncIntervalMillis = 60000 # Max records to fetch from Kinesis in a single GetRecords call. #maxRecords = 10000 # Idle time between record reads in milliseconds. #idleTimeBetweenReadsInMillis = 1000 # Enables applications flush/checkpoint (if they have some data "in progress", but don't get new data for while) #callProcessRecordsEvenForEmptyRecordList = false # Interval in milliseconds between polling to check for parent shard completion. # Polling frequently will take up more DynamoDB IOPS (when there are leases for shards waiting on # completion of parent shards). #parentShardPollIntervalMillis = 10000 # Cleanup leases upon shards completion (don't wait until they expire in Kinesis). # Keeping leases takes some tracking/resources (e.g. they need to be renewed, assigned), so by default we try # to delete the ones we don't need any longer. #cleanupLeasesUponShardCompletion = true # Backoff time in milliseconds for Amazon Kinesis Client Library tasks (in the event of failures). #taskBackoffTimeMillis = 500 # Buffer metrics for at most this long before publishing to CloudWatch. #metricsBufferTimeMillis = 10000 # Buffer at most this many metrics before publishing to CloudWatch. #metricsMaxQueueSize = 10000 # KCL will validate client provided sequence numbers with a call to Amazon Kinesis before checkpointing for calls # to RecordProcessorCheckpointer#checkpoint(String) by default. #validateSequenceNumberBeforeCheckpointing = true # The maximum number of active threads for the MultiLangDaemon to permit. # If a value is provided then a FixedThreadPool is used with the maximum # active threads set to the provided value. If a non-positive integer or no # value is provided a CachedThreadPool is used. #maxActiveThreads = 0 ``` ### Nombre de la aplicación KCL requiere un nombre de aplicación que sea único entre las aplicaciones y en las tablas de Amazon DynamoDB de la misma región. La biblioteca utiliza el valor del nombre de la aplicación de las siguientes formas: + Se supone que los procesos de trabajo que están asociados a este nombre de aplicación operan de forma conjunta en la misma secuencia. Estos procesos de trabajo pueden distribuirse entre varias instancias. Si ejecuta otra instancia del mismo código de aplicación, pero con otro nombre de aplicación, KCL considera que la segunda instancia es una aplicación completamente independiente de la otra que opera en el mismo flujo. + KCL crea una tabla de DynamoDB con el nombre de la aplicación y utiliza la tabla para actualizar la información de estado (como los puntos de verificación y el mapeo procesos de trabajo-particiones) para la aplicación. Cada aplicación tiene su propia tabla de DynamoDB. Para obtener más información, consulte [Usar una tabla de arrendamiento para realizar el seguimiento de las particiones procesadas por la aplicación de consumo de KCL](shared-throughput-kcl-consumers.md#shared-throughput-kcl-consumers-leasetable). ### Credenciales Debe poner sus AWS credenciales a disposición de uno de los proveedores de credenciales de la cadena de proveedores de [credenciales predeterminada](https://docs.aws.amazon.com/sdk-for-java/latest/reference/com/amazonaws/auth/DefaultAWSCredentialsProviderChain.html). Puede usar la propiedad `AWSCredentialsProvider` para configurar un proveedor de credenciales. Si ejecuta su aplicación de consumidor en una instancia de Amazon EC2, le recomendamos que configure la instancia con un rol de IAM. AWS Las credenciales que reflejan los permisos asociados a esta función de IAM se ponen a disposición de las aplicaciones de la instancia a través de los metadatos de la instancia. Esta es la forma más segura de administrar las credenciales para una aplicación consumidora que se ejecute en una instancia EC2. # Desarrollar consumidores de distribución mejorados con KCL 2.x **importante** Las versiones 1.x y 2.x de la biblioteca de clientes de Amazon Kinesis (KCL) están obsoletas. KCL 1.x estará disponible el 30 de end-of-support enero de 2026. **Recomendamos** que migre las aplicaciones de KCL que utilizan la versión 1.x a la última versión de KCL antes del 30 de enero de 2026. Para encontrar la versión más reciente de KCL, consulte la página de la [biblioteca de clientes de Amazon Kinesis](https://github.com/awslabs/amazon-kinesis-client) en. GitHub Para obtener información sobre las versiones más recientes de KCL, consulte [Uso de Kinesis Client Library](kcl.md). Para obtener información sobre cómo migrar de KCL 1.x a KCL 3.x, consulte [Migración de KCL 1.x a KCL 3.x](kcl-migration-1-3.md). Los consumidores que utilizan la *distribución ramificada mejorada* en Amazon Kinesis Data Streams pueden recibir los registros de un flujo de datos con un rendimiento dedicado de hasta 2 MB de datos por segundo por partición. Este tipo de consumidor no tiene que competir con otros consumidores que reciben datos de la secuencia. Para obtener más información, consulte [Desarrollo de consumidores de distribución ramificada mejorada con rendimiento dedicado](enhanced-consumers.md). Puede utilizar la versión 2.0 o posterior de Kinesis Client Library (KCL) para desarrollar aplicaciones que utilicen la distribución ramificada mejorada para recibir datos de los flujos. El KCL suscribe automáticamente su aplicación a todos los fragmentos de una transmisión y garantiza que su aplicación de consumo pueda leer con un valor de rendimiento de 2 por fragmento. MB/sec Si quiere utilizar KCL sin activar la distribución ramificada mejorada, consulte la página sobre [desarrollo de consumidores mediante Kinesis Client Library 2.0](https://docs.aws.amazon.com/streams/latest/dev/developing-consumers-with-kcl-v2.html). **Topics** + [ # Desarrollar consumidores de distribución mejorados con KCL 2.x en Java ](building-enhanced-consumers-kcl-java.md) # Desarrollar consumidores de distribución mejorados con KCL 2.x en Java **importante** Las versiones 1.x y 2.x de la biblioteca de clientes de Amazon Kinesis (KCL) están obsoletas. KCL 1.x llegará el 30 de enero de 2026. end-of-support **Recomendamos** que migre las aplicaciones de KCL que utilizan la versión 1.x a la última versión de KCL antes del 30 de enero de 2026. Para encontrar la versión más reciente de KCL, consulte la página de la [biblioteca de clientes de Amazon Kinesis](https://github.com/awslabs/amazon-kinesis-client) en. GitHub Para obtener información sobre las versiones más recientes de KCL, consulte [Uso de Kinesis Client Library](kcl.md). Para obtener información sobre cómo migrar de KCL 1.x a KCL 3.x, consulte [Migración de KCL 1.x a KCL 3.x](kcl-migration-1-3.md). Puede utilizar la versión 2.0 o posterior de Kinesis Client Library (KCL) para desarrollar aplicaciones en Amazon Kinesis Data Streams para recibir datos de los flujos mediante la distribución ramificada mejorada. En el siguiente código, se muestra un ejemplo de implementación en Java de `ProcessorFactory` y `RecordProcessor`. Es recomendable que utilice `KinesisClientUtil` para crear `KinesisAsyncClient` y para establecer `maxConcurrency` en `KinesisAsyncClient`. **importante** El cliente de Amazon Kinesis puede experimentar un importante aumento de la latencia, a menos que configure `KinesisAsyncClient` de forma que el valor de `maxConcurrency` sea lo suficientemente alto para permitir todas las asignaciones, además de los usos adicionales de `KinesisAsyncClient`. ``` /* * Copyright 2019 Amazon.com, Inc. or its affiliates. All Rights Reserved. * * Licensed under the Amazon Software License (the "License"). * You may not use this file except in compliance with the License. * A copy of the License is located at * * http://aws.amazon.com/asl/ * * or in the "license" file accompanying this file. This file is distributed * on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either * express or implied. See the License for the specific language governing * permissions and limitations under the License. */ /* * Copyright 2019 Amazon.com, Inc. or its affiliates. All Rights Reserved. * * Licensed under the Apache License, Version 2.0 (the "License"). * You may not use this file except in compliance with the License. * A copy of the License is located at * * http://www.apache.org/licenses/LICENSE-2.0 * * or in the "license" file accompanying this file. This file is distributed * on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either * express or implied. See the License for the specific language governing * permissions and limitations under the License. */ import java.io.BufferedReader; import java.io.IOException; import java.io.InputStreamReader; import java.util.UUID; import java.util.concurrent.ExecutionException; import java.util.concurrent.Executors; import java.util.concurrent.Future; import java.util.concurrent.ScheduledExecutorService; import java.util.concurrent.ScheduledFuture; import java.util.concurrent.TimeUnit; import java.util.concurrent.TimeoutException; import org.apache.commons.lang3.ObjectUtils; import org.apache.commons.lang3.RandomStringUtils; import org.apache.commons.lang3.RandomUtils; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.slf4j.MDC; import software.amazon.awssdk.core.SdkBytes; import software.amazon.awssdk.regions.Region; import software.amazon.awssdk.services.cloudwatch.CloudWatchAsyncClient; import software.amazon.awssdk.services.dynamodb.DynamoDbAsyncClient; import software.amazon.awssdk.services.kinesis.KinesisAsyncClient; import software.amazon.awssdk.services.kinesis.model.PutRecordRequest; import software.amazon.kinesis.common.ConfigsBuilder; import software.amazon.kinesis.common.KinesisClientUtil; import software.amazon.kinesis.coordinator.Scheduler; import software.amazon.kinesis.exceptions.InvalidStateException; import software.amazon.kinesis.exceptions.ShutdownException; import software.amazon.kinesis.lifecycle.events.InitializationInput; import software.amazon.kinesis.lifecycle.events.LeaseLostInput; import software.amazon.kinesis.lifecycle.events.ProcessRecordsInput; import software.amazon.kinesis.lifecycle.events.ShardEndedInput; import software.amazon.kinesis.lifecycle.events.ShutdownRequestedInput; import software.amazon.kinesis.processor.ShardRecordProcessor; import software.amazon.kinesis.processor.ShardRecordProcessorFactory; public class SampleSingle { private static final Logger log = LoggerFactory.getLogger(SampleSingle.class); public static void main(String... args) { if (args.length < 1) { log.error("At a minimum, the stream name is required as the first argument. The Region may be specified as the second argument."); System.exit(1); } String streamName = args[0]; String region = null; if (args.length > 1) { region = args[1]; } new SampleSingle(streamName, region).run(); } private final String streamName; private final Region region; private final KinesisAsyncClient kinesisClient; private SampleSingle(String streamName, String region) { this.streamName = streamName; this.region = Region.of(ObjectUtils.firstNonNull(region, "us-east-2")); this.kinesisClient = KinesisClientUtil.createKinesisAsyncClient(KinesisAsyncClient.builder().region(this.region)); } private void run() { ScheduledExecutorService producerExecutor = Executors.newSingleThreadScheduledExecutor(); ScheduledFuture producerFuture = producerExecutor.scheduleAtFixedRate(this::publishRecord, 10, 1, TimeUnit.SECONDS); DynamoDbAsyncClient dynamoClient = DynamoDbAsyncClient.builder().region(region).build(); CloudWatchAsyncClient cloudWatchClient = CloudWatchAsyncClient.builder().region(region).build(); ConfigsBuilder configsBuilder = new ConfigsBuilder(streamName, streamName, kinesisClient, dynamoClient, cloudWatchClient, UUID.randomUUID().toString(), new SampleRecordProcessorFactory()); Scheduler scheduler = new Scheduler( configsBuilder.checkpointConfig(), configsBuilder.coordinatorConfig(), configsBuilder.leaseManagementConfig(), configsBuilder.lifecycleConfig(), configsBuilder.metricsConfig(), configsBuilder.processorConfig(), configsBuilder.retrievalConfig() ); Thread schedulerThread = new Thread(scheduler); schedulerThread.setDaemon(true); schedulerThread.start(); System.out.println("Press enter to shutdown"); BufferedReader reader = new BufferedReader(new InputStreamReader(System.in)); try { reader.readLine(); } catch (IOException ioex) { log.error("Caught exception while waiting for confirm. Shutting down.", ioex); } log.info("Cancelling producer, and shutting down executor."); producerFuture.cancel(true); producerExecutor.shutdownNow(); Future gracefulShutdownFuture = scheduler.startGracefulShutdown(); log.info("Waiting up to 20 seconds for shutdown to complete."); try { gracefulShutdownFuture.get(20, TimeUnit.SECONDS); } catch (InterruptedException e) { log.info("Interrupted while waiting for graceful shutdown. Continuing."); } catch (ExecutionException e) { log.error("Exception while executing graceful shutdown.", e); } catch (TimeoutException e) { log.error("Timeout while waiting for shutdown. Scheduler may not have exited."); } log.info("Completed, shutting down now."); } private void publishRecord() { PutRecordRequest request = PutRecordRequest.builder() .partitionKey(RandomStringUtils.randomAlphabetic(5, 20)) .streamName(streamName) .data(SdkBytes.fromByteArray(RandomUtils.nextBytes(10))) .build(); try { kinesisClient.putRecord(request).get(); } catch (InterruptedException e) { log.info("Interrupted, assuming shutdown."); } catch (ExecutionException e) { log.error("Exception while sending data to Kinesis. Will try again next cycle.", e); } } private static class SampleRecordProcessorFactory implements ShardRecordProcessorFactory { public ShardRecordProcessor shardRecordProcessor() { return new SampleRecordProcessor(); } } private static class SampleRecordProcessor implements ShardRecordProcessor { private static final String SHARD_ID_MDC_KEY = "ShardId"; private static final Logger log = LoggerFactory.getLogger(SampleRecordProcessor.class); private String shardId; public void initialize(InitializationInput initializationInput) { shardId = initializationInput.shardId(); MDC.put(SHARD_ID_MDC_KEY, shardId); try { log.info("Initializing @ Sequence: {}", initializationInput.extendedSequenceNumber()); } finally { MDC.remove(SHARD_ID_MDC_KEY); } } public void processRecords(ProcessRecordsInput processRecordsInput) { MDC.put(SHARD_ID_MDC_KEY, shardId); try { log.info("Processing {} record(s)", processRecordsInput.records().size()); processRecordsInput.records().forEach(r -> log.info("Processing record pk: {} -- Seq: {}", r.partitionKey(), r.sequenceNumber())); } catch (Throwable t) { log.error("Caught throwable while processing records. Aborting."); Runtime.getRuntime().halt(1); } finally { MDC.remove(SHARD_ID_MDC_KEY); } } public void leaseLost(LeaseLostInput leaseLostInput) { MDC.put(SHARD_ID_MDC_KEY, shardId); try { log.info("Lost lease, so terminating."); } finally { MDC.remove(SHARD_ID_MDC_KEY); } } public void shardEnded(ShardEndedInput shardEndedInput) { MDC.put(SHARD_ID_MDC_KEY, shardId); try { log.info("Reached shard end checkpointing."); shardEndedInput.checkpointer().checkpoint(); } catch (ShutdownException | InvalidStateException e) { log.error("Exception while checkpointing at shard end. Giving up.", e); } finally { MDC.remove(SHARD_ID_MDC_KEY); } } public void shutdownRequested(ShutdownRequestedInput shutdownRequestedInput) { MDC.put(SHARD_ID_MDC_KEY, shardId); try { log.info("Scheduler is shutting down, checkpointing."); shutdownRequestedInput.checkpointer().checkpoint(); } catch (ShutdownException | InvalidStateException e) { log.error("Exception while checkpointing at requested shutdown. Giving up.", e); } finally { MDC.remove(SHARD_ID_MDC_KEY); } } } } ``` # Migrar consumidores de KCL 1.x a KCL 2.x **importante** Las versiones 1.x y 2.x de la biblioteca de clientes de Amazon Kinesis (KCL) están obsoletas. La versión 1.x de KCL estará disponible el 30 de end-of-support enero de 2026. **Recomendamos** que migre las aplicaciones de KCL que utilizan la versión 1.x a la última versión de KCL antes del 30 de enero de 2026. Para encontrar la versión más reciente de KCL, consulte la página de la [biblioteca de clientes de Amazon Kinesis](https://github.com/awslabs/amazon-kinesis-client) en. GitHub Para obtener información sobre las versiones más recientes de KCL, consulte [Uso de Kinesis Client Library](kcl.md). Para obtener información sobre cómo migrar de KCL 1.x a KCL 3.x, consulte [Migración de KCL 1.x a KCL 3.x](kcl-migration-1-3.md). En este tema se explican las diferencias entre las versiones 1.x y 2.x de Kinesis Client Library (KCL). También le muestra cómo migrar el consumidor de la versión 1.x a la versión 2.x de KCL. Después de migrar su cliente, comenzará el procesamiento de registros a partir de la última ubicación del punto de comprobación. La versión 2.0 de KCL presenta los siguientes cambios en la interfaz: **Cambios en la interfaz de KCL** | Interfaz de KCL 1.x | Interfaz de KCL 2.0 | | --- | --- | | com.amazonaws.services.kinesis.clientlibrary.interfaces.v2.IRecordProcessor | software.amazon.kinesis.processor.ShardRecordProcessor | | com.amazonaws.services.kinesis.clientlibrary.interfaces.v2.IRecordProcessorFactory | software.amazon.kinesis.processor.ShardRecordProcessorFactory | | com.amazonaws.services.kinesis.clientlibrary.interfaces.v2.IShutdownNotificationAware | Plegado en software.amazon.kinesis.processor.ShardRecordProcessor | **Topics** + [ ## Migrar el procesador de registros ](#recrod-processor-migration) + [ ## Migrar el generador de procesadores de registros ](#recrod-processor-factory-migration) + [ ## Migrar el proceso de trabajo ](#worker-migration) + [ ## Configurar el cliente de Amazon Kinesis ](#client-configuration) + [ ## Eliminar el tiempo de inactividad ](#idle-time-removal) + [ ## Eliminar la configuración de clientes ](#client-configuration-removals) ## Migrar el procesador de registros En el siguiente ejemplo, se muestra un procesador de registros implementado para la versión 1.x de KCL: ``` package com.amazonaws.kcl; import com.amazonaws.services.kinesis.clientlibrary.exceptions.InvalidStateException; import com.amazonaws.services.kinesis.clientlibrary.exceptions.ShutdownException; import com.amazonaws.services.kinesis.clientlibrary.interfaces.IRecordProcessorCheckpointer; import com.amazonaws.services.kinesis.clientlibrary.interfaces.v2.IRecordProcessor; import com.amazonaws.services.kinesis.clientlibrary.interfaces.v2.IShutdownNotificationAware; import com.amazonaws.services.kinesis.clientlibrary.lib.worker.ShutdownReason; import com.amazonaws.services.kinesis.clientlibrary.types.InitializationInput; import com.amazonaws.services.kinesis.clientlibrary.types.ProcessRecordsInput; import com.amazonaws.services.kinesis.clientlibrary.types.ShutdownInput; public class TestRecordProcessor implements IRecordProcessor, IShutdownNotificationAware { @Override public void initialize(InitializationInput initializationInput) { // // Setup record processor // } @Override public void processRecords(ProcessRecordsInput processRecordsInput) { // // Process records, and possibly checkpoint // } @Override public void shutdown(ShutdownInput shutdownInput) { if (shutdownInput.getShutdownReason() == ShutdownReason.TERMINATE) { try { shutdownInput.getCheckpointer().checkpoint(); } catch (ShutdownException | InvalidStateException e) { throw new RuntimeException(e); } } } @Override public void shutdownRequested(IRecordProcessorCheckpointer checkpointer) { try { checkpointer.checkpoint(); } catch (ShutdownException | InvalidStateException e) { // // Swallow exception // e.printStackTrace(); } } } ``` **Para migrar la clase del procesador de registros** 1. Cambie las interfaces de `com.amazonaws.services.kinesis.clientlibrary.interfaces.v2.IRecordProcessor` y `com.amazonaws.services.kinesis.clientlibrary.interfaces.v2.IShutdownNotificationAware` a `software.amazon.kinesis.processor.ShardRecordProcessor`, tal y como se indica a continuación: ``` // import com.amazonaws.services.kinesis.clientlibrary.interfaces.v2.IRecordProcessor; // import com.amazonaws.services.kinesis.clientlibrary.interfaces.v2.IShutdownNotificationAware; import software.amazon.kinesis.processor.ShardRecordProcessor; // public class TestRecordProcessor implements IRecordProcessor, IShutdownNotificationAware { public class TestRecordProcessor implements ShardRecordProcessor { ``` 1. Actualice las instrucciones `import` para los métodos `initialize` y `processRecords`. ``` // import com.amazonaws.services.kinesis.clientlibrary.types.InitializationInput; import software.amazon.kinesis.lifecycle.events.InitializationInput; //import com.amazonaws.services.kinesis.clientlibrary.types.ProcessRecordsInput; import software.amazon.kinesis.lifecycle.events.ProcessRecordsInput; ``` 1. Sustituya el método `shutdown` por los métodos nuevos siguientes: `leaseLost`, `shardEnded` y `shutdownRequested`. ``` // @Override // public void shutdownRequested(IRecordProcessorCheckpointer checkpointer) { // // // // This is moved to shardEnded(...) // // // try { // checkpointer.checkpoint(); // } catch (ShutdownException | InvalidStateException e) { // // // // Swallow exception // // // e.printStackTrace(); // } // } @Override public void leaseLost(LeaseLostInput leaseLostInput) { } @Override public void shardEnded(ShardEndedInput shardEndedInput) { try { shardEndedInput.checkpointer().checkpoint(); } catch (ShutdownException | InvalidStateException e) { // // Swallow the exception // e.printStackTrace(); } } // @Override // public void shutdownRequested(IRecordProcessorCheckpointer checkpointer) { // // // // This is moved to shutdownRequested(ShutdownReauestedInput) // // // try { // checkpointer.checkpoint(); // } catch (ShutdownException | InvalidStateException e) { // // // // Swallow exception // // // e.printStackTrace(); // } // } @Override public void shutdownRequested(ShutdownRequestedInput shutdownRequestedInput) { try { shutdownRequestedInput.checkpointer().checkpoint(); } catch (ShutdownException | InvalidStateException e) { // // Swallow the exception // e.printStackTrace(); } } ``` A continuación, se muestra la versión actualizada de la clase del procesador de registros. ``` package com.amazonaws.kcl; import software.amazon.kinesis.exceptions.InvalidStateException; import software.amazon.kinesis.exceptions.ShutdownException; import software.amazon.kinesis.lifecycle.events.InitializationInput; import software.amazon.kinesis.lifecycle.events.LeaseLostInput; import software.amazon.kinesis.lifecycle.events.ProcessRecordsInput; import software.amazon.kinesis.lifecycle.events.ShardEndedInput; import software.amazon.kinesis.lifecycle.events.ShutdownRequestedInput; import software.amazon.kinesis.processor.ShardRecordProcessor; public class TestRecordProcessor implements ShardRecordProcessor { @Override public void initialize(InitializationInput initializationInput) { } @Override public void processRecords(ProcessRecordsInput processRecordsInput) { } @Override public void leaseLost(LeaseLostInput leaseLostInput) { } @Override public void shardEnded(ShardEndedInput shardEndedInput) { try { shardEndedInput.checkpointer().checkpoint(); } catch (ShutdownException | InvalidStateException e) { // // Swallow the exception // e.printStackTrace(); } } @Override public void shutdownRequested(ShutdownRequestedInput shutdownRequestedInput) { try { shutdownRequestedInput.checkpointer().checkpoint(); } catch (ShutdownException | InvalidStateException e) { // // Swallow the exception // e.printStackTrace(); } } } ``` ## Migrar el generador de procesadores de registros El generador de procesadores de registros es responsable de la creación de procesadores de registros cuando se adquiere una asignación. A continuación, se muestra un ejemplo de un generador de la versión 1.x de KCL. ``` package com.amazonaws.kcl; import com.amazonaws.services.kinesis.clientlibrary.interfaces.v2.IRecordProcessor; import com.amazonaws.services.kinesis.clientlibrary.interfaces.v2.IRecordProcessorFactory; public class TestRecordProcessorFactory implements IRecordProcessorFactory { @Override public IRecordProcessor createProcessor() { return new TestRecordProcessor(); } } ``` **Para migrar el generador de procesadores de registros** 1. Cambie la interfaz implementada de `com.amazonaws.services.kinesis.clientlibrary.interfaces.v2.IRecordProcessorFactory` a `software.amazon.kinesis.processor.ShardRecordProcessorFactory`, tal y como se indica a continuación: ``` // import com.amazonaws.services.kinesis.clientlibrary.interfaces.v2.IRecordProcessor; import software.amazon.kinesis.processor.ShardRecordProcessor; // import com.amazonaws.services.kinesis.clientlibrary.interfaces.v2.IRecordProcessorFactory; import software.amazon.kinesis.processor.ShardRecordProcessorFactory; // public class TestRecordProcessorFactory implements IRecordProcessorFactory { public class TestRecordProcessorFactory implements ShardRecordProcessorFactory { ``` 1. Cambie la firma de retorno de `createProcessor`. ``` // public IRecordProcessor createProcessor() { public ShardRecordProcessor shardRecordProcessor() { ``` A continuación, se muestra un ejemplo de generador de procesadores de registros de la versión 2.0: ``` package com.amazonaws.kcl; import software.amazon.kinesis.processor.ShardRecordProcessor; import software.amazon.kinesis.processor.ShardRecordProcessorFactory; public class TestRecordProcessorFactory implements ShardRecordProcessorFactory { @Override public ShardRecordProcessor shardRecordProcessor() { return new TestRecordProcessor(); } } ``` ## Migrar el proceso de trabajo En la versión 2.0 de KCL, una nueva clase, llamada `Scheduler`, sustituye a la clase `Worker`. A continuación, se muestra un ejemplo de proceso de trabajo de la versión 1.x de KCL. ``` final KinesisClientLibConfiguration config = new KinesisClientLibConfiguration(...) final IRecordProcessorFactory recordProcessorFactory = new RecordProcessorFactory(); final Worker worker = new Worker.Builder() .recordProcessorFactory(recordProcessorFactory) .config(config) .build(); ``` **Para migrar el proceso de trabajo** 1. Cambie la instrucción `import` para la clase `Worker` por las instrucciones de importación para las clases `Scheduler` y `ConfigsBuilder`. ``` // import com.amazonaws.services.kinesis.clientlibrary.lib.worker.Worker; import software.amazon.kinesis.coordinator.Scheduler; import software.amazon.kinesis.common.ConfigsBuilder; ``` 1. Cree un `ConfigsBuilder` y un `Scheduler` como se muestra en el ejemplo siguiente. Es recomendable que utilice `KinesisClientUtil` para crear `KinesisAsyncClient` y para establecer `maxConcurrency` en `KinesisAsyncClient`. **importante** El cliente de Amazon Kinesis puede experimentar un importante aumento de la latencia, a menos que configure `KinesisAsyncClient` de forma que el valor de `maxConcurrency` sea lo suficientemente alto para permitir todas las asignaciones, además de los usos adicionales de `KinesisAsyncClient`. ``` import java.util.UUID; import software.amazon.awssdk.regions.Region; import software.amazon.awssdk.services.dynamodb.DynamoDbAsyncClient; import software.amazon.awssdk.services.cloudwatch.CloudWatchAsyncClient; import software.amazon.awssdk.services.kinesis.KinesisAsyncClient; import software.amazon.kinesis.common.ConfigsBuilder; import software.amazon.kinesis.common.KinesisClientUtil; import software.amazon.kinesis.coordinator.Scheduler; ... Region region = Region.AP_NORTHEAST_2; KinesisAsyncClient kinesisClient = KinesisClientUtil.createKinesisAsyncClient(KinesisAsyncClient.builder().region(region)); DynamoDbAsyncClient dynamoClient = DynamoDbAsyncClient.builder().region(region).build(); CloudWatchAsyncClient cloudWatchClient = CloudWatchAsyncClient.builder().region(region).build(); ConfigsBuilder configsBuilder = new ConfigsBuilder(streamName, applicationName, kinesisClient, dynamoClient, cloudWatchClient, UUID.randomUUID().toString(), new SampleRecordProcessorFactory()); Scheduler scheduler = new Scheduler( configsBuilder.checkpointConfig(), configsBuilder.coordinatorConfig(), configsBuilder.leaseManagementConfig(), configsBuilder.lifecycleConfig(), configsBuilder.metricsConfig(), configsBuilder.processorConfig(), configsBuilder.retrievalConfig() ); ``` ## Configurar el cliente de Amazon Kinesis Con el lanzamiento de la versión 2.0 de Kinesis Client Library, la configuración del cliente pasa de tener una única clase de configuración (`KinesisClientLibConfiguration`) a tener seis clases de configuración. En la tabla siguiente, se describe la migración. **Campos de configuración y sus clases nuevas** | Campo original | Clase de configuración nueva | Description (Descripción) | | --- | --- | --- | | applicationName | ConfigsBuilder | El nombre de la aplicación de KCL. Se utiliza de forma predeterminada para tableName y consumerName. | | tableName | ConfigsBuilder | Permite sustituir el nombre de la tabla que se utiliza para la tabla de asignaciones de Amazon DynamoDB. | | streamName | ConfigsBuilder | El nombre de la secuencia cuyos registros procesa de esta aplicación. | | kinesisEndpoint | ConfigsBuilder | Esta opción se ha eliminado. Consulte Eliminación de los parámetros de configuración de clientes. | | dynamoDBEndpoint | ConfigsBuilder | Esta opción se ha eliminado. Consulte Eliminación de los parámetros de configuración de clientes. | | initialPositionInStreamExtended | RetrievalConfig | La ubicación de la partición desde la que KCL comienza a recuperar registros, empezando por la ejecución inicial de la aplicación. | | kinesisCredentialsProvider | ConfigsBuilder | Esta opción se ha eliminado. Consulte Eliminación de los parámetros de configuración de clientes. | | dynamoDBCredentialsProvider | ConfigsBuilder | Esta opción se ha eliminado. Consulte Eliminación de los parámetros de configuración de clientes. | | cloudWatchCredentialsProvider | ConfigsBuilder | Esta opción se ha eliminado. Consulte Eliminación de los parámetros de configuración de clientes. | | failoverTimeMillis | LeaseManagementConfig | El número de milisegundos que deben transcurrir antes de que se pueda considerar que se ha producido un error en el propietario de una asignación. | | workerIdentifier | ConfigsBuilder | Identificador único que representa esta instancia del procesador de aplicaciones. Deben ser único. | | shardSyncIntervalMillis | LeaseManagementConfig | El tiempo entre llamadas de sincronización del fragmento. | | maxRecords | PollingConfig | Permite establecer el número máximo de registros que devuelve Kinesis. | | idleTimeBetweenReadsInMillis | CoordinatorConfig | Esta opción se ha eliminado. Consulte Eliminación del tiempo de inactividad. | | callProcessRecordsEvenForEmptyRecordList | ProcessorConfig | Cuando se establece, se llama al procesador de registros incluso cuando no se proporciona ningún registro de Kinesis. | | parentShardPollIntervalMillis | CoordinatorConfig | Determina la frecuencia con que debería sondear un procesador de registros para ver si el fragmento principal se ha completado. | | cleanupLeasesUponShardCompletion | LeaseManagementConfig | Cuando se establece, se eliminan las asignaciones tan pronto como se inicia el procesamiento del fragmento secundario. | | ignoreUnexpectedChildShards | LeaseManagementConfig | Cuando se establece, los fragmentos secundarios que tienen un fragmento abierto se pasan por alto. Esto es principalmente para DynamoDB Streams. | | kinesisClientConfig | ConfigsBuilder | Esta opción se ha eliminado. Consulte Eliminación de los parámetros de configuración de clientes. | | dynamoDBClientConfig | ConfigsBuilder | Esta opción se ha eliminado. Consulte Eliminación de los parámetros de configuración de clientes. | | cloudWatchClientConfig | ConfigsBuilder | Esta opción se ha eliminado. Consulte Eliminación de los parámetros de configuración de clientes. | | taskBackoffTimeMillis | LifecycleConfig | El tiempo que se debe esperar para reintentar la tareas con errores. | | metricsBufferTimeMillis | MetricsConfig | Controla la publicación de CloudWatch métricas. | | metricsMaxQueueSize | MetricsConfig | Controla la publicación de CloudWatch métricas. | | metricsLevel | MetricsConfig | Controla la publicación de CloudWatch métricas. | | metricsEnabledDimensions | MetricsConfig | Controla la publicación de CloudWatch métricas. | | validateSequenceNumberBeforeCheckpointing | CheckpointConfig | Esta opción se ha eliminado. Consulte Validación del número de secuencia del punto de comprobación. | | regionName | ConfigsBuilder | Esta opción se ha eliminado. Consulte Eliminación de los parámetros de configuración de clientes. | | maxLeasesForWorker | LeaseManagementConfig | El número máximo de asignaciones que debería aceptar una sola instancia de la aplicación. | | maxLeasesToStealAtOneTime | LeaseManagementConfig | El número máximo de asignaciones del que debería intentar apropiarse una aplicación al mismo tiempo. | | initialLeaseTableReadCapacity | LeaseManagementConfig | La IOPs lectura de DynamoDB que se utiliza si la biblioteca de clientes de Kinesis necesita crear una nueva tabla de concesiones de DynamoDB. | | initialLeaseTableWriteCapacity | LeaseManagementConfig | La IOPs lectura de DynamoDB que se utiliza si la biblioteca de clientes de Kinesis necesita crear una nueva tabla de concesiones de DynamoDB. | | initialPositionInStreamExtended | LeaseManagementConfig | La posición inicial de la secuencia en la que debería comenzar la aplicación. Esto solo se utiliza durante la creación inicial de la asignación. | | skipShardSyncAtWorkerInitializationIfLeasesExist | CoordinatorConfig | Deshabilita la sincronización de los datos de los fragmentos si la tabla de asignaciones todavía contiene entradas. KinesisEcoTODO: -438 | | shardPrioritization | CoordinatorConfig | La priorización de fragmentos que se va a utilizar. | | shutdownGraceMillis | N/A | Esta opción se ha eliminado. Consulte MultiLang Mudanzas. | | timeoutInSeconds | N/A | Esta opción se ha eliminado. Consulte MultiLang Mudanzas. | | retryGetRecordsInSeconds | PollingConfig | Configura el retraso entre GetRecords intentos en caso de error. | | maxGetRecordsThreadPool | PollingConfig | El tamaño del grupo de subprocesos utilizado para GetRecords. | | maxLeaseRenewalThreads | LeaseManagementConfig | Controla el tamaño del grupo de subprocesos del renovador de asignaciones. Cuanto más grande sea el número de asignaciones que puede tomar la aplicación, más grande debe ser este grupo. | | recordsFetcherFactory | PollingConfig | Permite sustituir el generador que se utiliza para crear capturadores que recuperan datos de las secuencias. | | logWarningForTaskAfterMillis | LifecycleConfig | Tiempo que se debe esperar antes de registrar una advertencia si una tarea no ha finalizado. | | listShardsBackoffTimeInMillis | RetrievalConfig | El número de milisegundos que se debe esperar entre llamadas a ListShards cuando se producen errores. | | maxListShardsRetryAttempts | RetrievalConfig | El número máximo de veces que se reintenta ListShards antes de desistir. | ## Eliminar el tiempo de inactividad En la versión 1.x de KCL, `idleTimeBetweenReadsInMillis` correspondía a dos cantidades: + La cantidad de tiempo entre envíos de tareas. Ahora puede configurar este tiempo entre tareas estableciendo `CoordinatorConfig#shardConsumerDispatchPollIntervalMillis`. + La cantidad de tiempo en reposo cuando no se devuelven registros desde Kinesis Data Streams. En la versión 2.0, con la distribución ramificada mejorada, los registros se envían desde sus respectivos recuperadores. Solo se produce actividad en el consumidor del fragmento cuando se recibe una solicitud enviada. ## Eliminar la configuración de clientes En la versión 2.0, KCL ya no crea los clientes. El usuario es el responsable de suministrar un cliente válido. Con este cambio, se han eliminado todos los parámetros de configuración que controlaban la creación de clientes. Si necesita estos parámetros, puede establecerlos en los clientes antes de proporcionar clientes a `ConfigsBuilder`. **** | Campo eliminado | Configuración equivalente | | --- | --- | | kinesisEndpoint | Configure el SDK KinesisAsyncClient con el punto de enlace preferido: KinesisAsyncClient.builder().endpointOverride(URI.create("https://")).build(). | | dynamoDBEndpoint | Configure el SDK DynamoDbAsyncClient con el punto de enlace preferido: DynamoDbAsyncClient.builder().endpointOverride(URI.create("https://")).build(). | | kinesisClientConfig | Configure el SDK KinesisAsyncClient con la configuración necesaria: KinesisAsyncClient.builder().overrideConfiguration().build(). | | dynamoDBClientConfig | Configure el SDK DynamoDbAsyncClient con la configuración necesaria: DynamoDbAsyncClient.builder().overrideConfiguration().build(). | | cloudWatchClientConfig | Configure el SDK CloudWatchAsyncClient con la configuración necesaria: CloudWatchAsyncClient.builder().overrideConfiguration().build(). | | regionName | Configure el SDK con la región preferida. Es la misma para todos los clientes del SDK. Por ejemplo, KinesisAsyncClient.builder().region(Region.US\$1WEST\$12).build(). | # Desarrolle a los consumidores con AWS SDK para Java Puede desarrollar consumidores personalizados mediante Amazon Kinesis Data APIs Streams. En esta sección se describe el uso de Kinesis Data APIs Streams con AWS SDK para Java. **importante** El método recomendado para desarrollar consumidores personalizados de Kinesis Data Streams con rendimiento compartido es utilizar la Bibliotecta de clientes de Kinesis (KCL). KCL ayuda a consumir y procesar los datos de un flujo de datos de Kinesis, ya que se encarga de muchas de las tareas complejas asociadas a la computación distribuida. Para obtener más información, consulte [Desarrollar consumidores con KCL en Java](develop-kcl-consumers-java.md). **Topics** + [ # Desarrolle consumidores con un rendimiento compartido con AWS SDK para Java ](developing-consumers-with-sdk.md) + [ # Desarrolle un mayor número de consumidores con el AWS SDK para Java ](building-enhanced-consumers-api.md) + [ # Interactúe con los datos mediante el registro de esquemas AWS Glue ](building-enhanced-consumers-glue-schema-registry.md) # Desarrolle consumidores con un rendimiento compartido con AWS SDK para Java Uno de los métodos para desarrollar Kinesis Data Streams personalizados para los consumidores con contenido compartido en todo momento consiste en utilizar Amazon Kinesis Data APIs Streams con. AWS SDK para Java En esta sección se describe el uso de Kinesis Data APIs Streams con AWS SDK para Java. Puede llamar a Kinesis Data APIs Streams mediante otros lenguajes de programación diferentes. 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/). El código de muestra de Java de esta sección indica 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ódigo listo para producción. No comprueban todas las excepciones posibles ni tienen en cuenta todas las consideraciones de seguridad y de rendimiento. **Topics** + [ ## Obtener datos de un flujo ](#kinesis-using-sdk-java-get-data) + [ ## Usar iteradores de particiones ](#kinesis-using-sdk-java-get-data-shard-iterators) + [ ## Utilice GetRecords ](#kinesis-using-sdk-java-get-data-getrecords) + [ ## Adaptarse a una nueva partición ](#kinesis-using-sdk-java-get-data-reshard) ## Obtener datos de un flujo Los Kinesis Data APIs Streams incluyen `getShardIterator` los métodos `getRecords` y que puede invocar para recuperar registros de un flujo de datos. Se trata del modelo de extracción, donde el código extrae registros de datos directamente de las particiones del flujo de datos. **importante** Se recomienda utilizar la funcionalidad de procesador de registros que proporciona KCL para recuperarlos de los flujos de datos. Se trata del modelo de inserción, en el que debe implementar el código que procesa los datos. KCL recupera los registros de datos del flujo de datos y los entrega al código de la aplicación. Además, KCL proporciona funciones de recuperación, conmutación por error y equilibrio de carga. Para obtener más información, consulte [Desarrollo de consumidores personalizados con rendimiento compartido mediante KCL](https://docs.aws.amazon.com/streams/latest/dev/shared-throughput-kcl-consumers.html). Sin embargo, en algunos casos puede que prefiera utilizar Kinesis Data APIs Streams. Por ejemplo, para implementar herramientas personalizadas para la supervisión o la depuración de los flujos de datos. **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). ## Usar iteradores de particiones Puede recuperar registros desde la secuencia por fragmentos. Para cada fragmento y cada lote de registros obtenido de ese fragmento debe conseguir un *iterador de fragmentos*. El iterador de fragmentos se utiliza en el objeto `getRecordsRequest` para especificar el fragmento a partir del cual deben recuperarse los registros. El tipo asociado con el iterador de fragmentos determina el punto del fragmento a partir del cual deben recuperarse los registros (consulte la información que se incluye más adelante en esta sección para obtener más detalles). Antes de trabajar con el iterador de particiones, tendrá que recuperar la partición. Para obtener más información, consulte [Obtener lista de particiones](kinesis-using-sdk-java-list-shards.md). Obtenga el iterador de fragmentos inicial con el método `getShardIterator`. Obtenga iteradores de fragmentos para lotes adicionales de registros utilizando el método `getNextShardIterator` del objeto `getRecordsResult` que devuelve el método `getRecords`. Un iterador de fragmentos es válido durante 5 minutos. Si utiliza un iterador de fragmentos mientras sea válido, obtendrá uno nuevo. Cada iterador de fragmentos mantiene su validez durante 5 minutos, incluso después de utilizarlo. Para obtener el iterador de fragmentos inicial, cree instancias de `GetShardIteratorRequest` y páselas al método `getShardIterator`. Para configurar la solicitud, especifique la secuencia y el ID del fragmento. Para obtener información sobre cómo obtener las transmisiones de su AWS cuenta, consulte[Lista de secuencias](kinesis-using-sdk-java-list-streams.md). Para obtener información sobre cómo obtener los fragmentos en una secuencia, consulte [Obtener lista de particiones](kinesis-using-sdk-java-list-shards.md). ``` String shardIterator; GetShardIteratorRequest getShardIteratorRequest = new GetShardIteratorRequest(); getShardIteratorRequest.setStreamName(myStreamName); getShardIteratorRequest.setShardId(shard.getShardId()); getShardIteratorRequest.setShardIteratorType("TRIM_HORIZON"); GetShardIteratorResult getShardIteratorResult = client.getShardIterator(getShardIteratorRequest); shardIterator = getShardIteratorResult.getShardIterator(); ``` Este código de muestra especifica `TRIM_HORIZON` como el tipo de iterador que se utiliza para obtener el iterador de fragmentos inicial. Este tipo de iterador implica que se deben devolver los registros y comenzar por el primer registro agregado a la partición, en lugar de comenzar por el registro agregado más recientemente, también denominado *extremo*. Los tipos de iteradores posibles son los siguientes: + `AT_SEQUENCE_NUMBER` + `AFTER_SEQUENCE_NUMBER` + `AT_TIMESTAMP` + `TRIM_HORIZON` + `LATEST` Para obtener más información, consulte [ShardIteratorType](https://docs.aws.amazon.com/kinesis/latest/APIReference/API_GetShardIterator.html#Kinesis-GetShardIterator-request-ShardIteratorType). Algunos tipos de iteradores requieren que se especifique un número de secuencia además del tipo, por ejemplo: ``` getShardIteratorRequest.setShardIteratorType("AT_SEQUENCE_NUMBER"); getShardIteratorRequest.setStartingSequenceNumber(specialSequenceNumber); ``` Después de obtener un registro mediante `getRecords`, puede obtener el número de secuencia del registro si llama al método `getSequenceNumber` del registro. ``` record.getSequenceNumber() ``` Además, el código que añade registros a la secuencia de datos puede obtener el número de secuencia para un registro añadido llamando a `getSequenceNumber` en el resultado de `putRecord`. ``` lastSequenceNumber = putRecordResult.getSequenceNumber(); ``` Puede utilizar números secuenciales para garantizar que los registros tengan un orden estrictamente ascendente. Para obtener más información, consulte el código de ejemplo en [PutRecord ejemplo](developing-producers-with-sdk.md#kinesis-using-sdk-java-putrecord-example). ## Utilice GetRecords Una vez que haya obtenido el iterador de fragmentos, cree una instancia de un objeto `GetRecordsRequest`. Especifique el iterador para la solicitud con el método `setShardIterator`. También puede establecer el número de registros que quiera recuperar mediante el método `setLimit`. El número de registros que devuelve `getRecords` es siempre igual o inferior a este límite. Si no especifica este límite, `getRecords` devuelve 10 MB de registros recuperados. El código de muestra que aparece a continuación establece este límite en 25 registros. Si no se devuelven, significa que no hay registros de datos disponibles actualmente en este fragmento con el número de secuencia al que hace referencia el iterador de fragmentos. En una situación así, la aplicación debe esperar una cantidad de tiempo adecuada para los orígenes de datos del flujo. Intente obtener datos de nuevo a partir del fragmento mediante el iterador de fragmentos que ha devuelto la llamada anterior a `getRecords`. Pase la `getRecordsRequest` al método `getRecords` y capture el valor devuelto como un objeto `getRecordsResult`. Para obtener los registros de datos, llame al método `getRecords` en el objeto `getRecordsResult`. ``` GetRecordsRequest getRecordsRequest = new GetRecordsRequest(); getRecordsRequest.setShardIterator(shardIterator); getRecordsRequest.setLimit(25); GetRecordsResult getRecordsResult = client.getRecords(getRecordsRequest); List records = getRecordsResult.getRecords(); ``` Para prepararse para otra llamada a `getRecords`, obtenga el siguiente iterador de fragmentos desde `getRecordsResult`. ``` shardIterator = getRecordsResult.getNextShardIterator(); ``` Para obtener resultados óptimos, suspenda la actividad durante al menos 1 segundo (1000 milisegundos) entre las llamadas a `getRecords` para evitar que se supere el límite de frecuencia de `getRecords`. ``` try { Thread.sleep(1000); } catch (InterruptedException e) {} ``` Normalmente, debe llamar a `getRecords` en bucle, incluso cuando recupere un solo registro en un entorno de pruebas. Una única llamada a `getRecords` podría devolver una lista de registros vacía, incluso si el fragmento contiene más registros en números secuenciales posteriores. Si ocurre esto, el `NextShardIterator` que se devuelve junto con la lista de registros vacía hace referencia a un número de secuencia posterior en el fragmento, y las llamadas sucesivas a `getRecords` acabarán por devolver los registros. El siguiente ejemplo ilustra el uso de un bucle. **Ejemplo: getRecords** El siguiente ejemplo de código refleja las sugerencias sobre `getRecords` que hemos planteado en esta sección, incluidas la realización de llamadas en bucle. ``` // Continuously read data records from a shard List records; while (true) { // Create a new getRecordsRequest with an existing shardIterator // Set the maximum records to return to 25 GetRecordsRequest getRecordsRequest = new GetRecordsRequest(); getRecordsRequest.setShardIterator(shardIterator); getRecordsRequest.setLimit(25); GetRecordsResult result = client.getRecords(getRecordsRequest); // Put the result into record list. The result can be empty. records = result.getRecords(); try { Thread.sleep(1000); } catch (InterruptedException exception) { throw new RuntimeException(exception); } shardIterator = result.getNextShardIterator(); } ``` Si utiliza Kinesis Client Library, esta podría hacer varias llamadas antes de devolver los datos. Este es el comportamiento intencionado según el diseño y no indica ningún problema con KCL o los datos. ## Adaptarse a una nueva partición Si `getRecordsResult.getNextShardIterator` devuelve `null`, indica que se ha producido una división o combinación de una partición que ha implicado esta partición. Esta partición se encuentra ahora en un estado `CLOSED` y se han leído todos los registros de datos disponibles de esta partición. En este escenario, puede utilizar `getRecordsResult.childShards` para obtener información sobre las nuevas particiones secundarias de la partición que se procesa y que se crearon mediante la división o la combinación. Para obtener más información, consulte [ChildShard](https://docs.aws.amazon.com/kinesis/latest/APIReference/API_ChildShard.html). En el caso de una división, los dos nuevos fragmentos tienen un `parentShardId` igual al ID de fragmento del fragmento que estuviera procesando anteriormente. El valor de `adjacentParentShardId` para ambos fragmentos es `null`. En el caso de una fusión, el único fragmento nuevo creado por la fusión tiene un `parentShardId` igual al ID del fragmento de uno de los fragmentos de origen y un `adjacentParentShardId` igual al ID de fragmento del otro fragmento de origen. La aplicación ya ha leído todos los datos de uno de estos fragmentos. Este es el fragmento para el que `getRecordsResult.getNextShardIterator` ha devuelto `null`. Si el orden de los datos es importante en la aplicación, debe asegurarse de que esta también lea todos los datos del otro fragmento principal antes de leer datos nuevos del fragmento secundario creado por la fusión. Si utiliza varios procesadores para recuperar los datos de la secuencia (por ejemplo, un procesador por fragmento) y se produce una división o fusión de fragmentos, debe aumentar o disminuir el número de procesadores para adaptarse a los cambios en el número de fragmentos. Para obtener más información acerca de cómo realizar cambios en los fragmentos, incluida una explicación de los estados de los fragmentos, como `CLOSED`, consulte [Cambiar las particiones de un flujo](kinesis-using-sdk-java-resharding.md). # Desarrolle un mayor número de consumidores con el AWS SDK para Java La *distribución ramificada mejorada* es una característica de Amazon Kinesis Data Streams que permite a los consumidores recibir registros de un flujo de datos con un rendimiento dedicado de hasta 2 MB de datos por segundo por partición. Un consumidor que utiliza la distribución ramificada mejorada no tiene que competir con otros consumidores que reciben datos de la secuencia. Para obtener más información, consulte [Desarrollo de consumidores de distribución ramificada mejorada con rendimiento dedicado](enhanced-consumers.md). Puede utilizar las operaciones de la API para crear un consumidor que utilice la distribución ramificada mejorada en Kinesis Data Streams. **Para registrar un consumidor con distribución ramificada mejorada mediante la API de Kinesis Data Streams** 1. Llame [RegisterStreamConsumer](https://docs.aws.amazon.com/kinesis/latest/APIReference/API_RegisterStreamConsumer.html)para registrar su solicitud como consumidor que utiliza un sistema de distribución ampliado. Kinesis Data Streams genera un nombre de recurso de Amazon (ARN) para el consumidor y lo devuelve en la respuesta. 1. Para empezar a escuchar un fragmento específico, pasa el ARN del consumidor en una llamada a. [SubscribeToShard](https://docs.aws.amazon.com/kinesis/latest/APIReference/API_SubscribeToShard.html) A continuación, Kinesis Data Streams comienza a enviarle los registros de ese fragmento, en forma de eventos de [SubscribeToShardEvent](https://docs.aws.amazon.com/kinesis/latest/APIReference/API_SubscribeToShardEvent.html)tipo a través de una conexión HTTP/2. La conexión permanece abierta durante un máximo de 5 minutos. [SubscribeToShard](https://docs.aws.amazon.com/kinesis/latest/APIReference/API_SubscribeToShard.html)Vuelva a llamar si quiere seguir recibiendo los registros del fragmento una vez `future` que la llamada devuelva y se [SubscribeToShard](https://docs.aws.amazon.com/kinesis/latest/APIReference/API_SubscribeToShard.html)complete de forma normal o excepcional. **nota** La API `SubscribeToShard` también devuelve la lista de las particiones secundarias de la partición actual cuando se alcanza el final de la partición actual. 1. Para anular el registro de un consumidor que utiliza la función de distribución mejorada, llama al teléfono. [DeregisterStreamConsumer](https://docs.aws.amazon.com/kinesis/latest/APIReference/API_DeregisterStreamConsumer.html) El código siguiente es un ejemplo de cómo suscribir el consumidor a un fragmento, renovar la suscripción de forma periódica y controlar los eventos. ``` import software.amazon.awssdk.services.kinesis.KinesisAsyncClient; import software.amazon.awssdk.services.kinesis.model.ShardIteratorType; import software.amazon.awssdk.services.kinesis.model.SubscribeToShardEvent; import software.amazon.awssdk.services.kinesis.model.SubscribeToShardRequest; import software.amazon.awssdk.services.kinesis.model.SubscribeToShardResponseHandler; import java.util.concurrent.CompletableFuture; /** * See https://github.com/awsdocs/aws-doc-sdk-examples/blob/master/javav2/example_code/kinesis/src/main/java/com/example/kinesis/KinesisStreamEx.java * for complete code and more examples. */ public class SubscribeToShardSimpleImpl { private static final String CONSUMER_ARN = "arn:aws:kinesis:us-east-1:123456789123:stream/foobar/consumer/test-consumer:1525898737"; private static final String SHARD_ID = "shardId-000000000000"; public static void main(String[] args) { KinesisAsyncClient client = KinesisAsyncClient.create(); SubscribeToShardRequest request = SubscribeToShardRequest.builder() .consumerARN(CONSUMER_ARN) .shardId(SHARD_ID) .startingPosition(s -> s.type(ShardIteratorType.LATEST)).build(); // Call SubscribeToShard iteratively to renew the subscription periodically. while(true) { // Wait for the CompletableFuture to complete normally or exceptionally. callSubscribeToShardWithVisitor(client, request).join(); } // Close the connection before exiting. // client.close(); } /** * Subscribes to the stream of events by implementing the SubscribeToShardResponseHandler.Visitor interface. */ private static CompletableFuture callSubscribeToShardWithVisitor(KinesisAsyncClient client, SubscribeToShardRequest request) { SubscribeToShardResponseHandler.Visitor visitor = new SubscribeToShardResponseHandler.Visitor() { @Override public void visit(SubscribeToShardEvent event) { System.out.println("Received subscribe to shard event " + event); } }; SubscribeToShardResponseHandler responseHandler = SubscribeToShardResponseHandler .builder() .onError(t -> System.err.println("Error during stream - " + t.getMessage())) .subscriber(visitor) .build(); return client.subscribeToShard(request, responseHandler); } } ``` Si `event.ContinuationSequenceNumber` devuelve `null`, indica que se ha producido una división o combinación de una partición que ha implicado esta partición. Esta partición se encuentra ahora en un estado `CLOSED`, y se han leído todos los registros de datos disponibles de esta partición. En este escenario, según el ejemplo anterior, puede utilizar `event.childShards` para obtener información sobre las nuevas particiones secundarias de la partición que se procesa y que se crearon mediante la división o la combinación. Para obtener más información, consulte [ChildShard](https://docs.aws.amazon.com/kinesis/latest/APIReference/API_ChildShard.html). # Interactúe con los datos mediante el registro de esquemas AWS Glue 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 formas de configurar esta integración es mediante la API de `GetRecords` Kinesis Data Streams, disponible en AWS el SDK de Java. Para obtener instrucciones detalladas sobre cómo configurar la integración de Kinesis Data Streams con Schema Registry mediante Kinesis Data APIs Streams, consulte `GetRecords` la [sección «Interacción con datos mediante Kinesis Data Streams» en Caso de uso: integración de Amazon Kinesis APIs Data Streams con el registro de Glue](https://docs.aws.amazon.com/glue/latest/dg/schema-registry-integrations.html#schema-registry-integrations-kds) Schema. AWS # Desarrolle a los consumidores utilizando AWS Lambda Puede utilizar una AWS Lambda función para procesar los registros de un flujo de datos. AWS Lambda es un servicio informático que permite ejecutar código sin aprovisionar ni administrar servidores. Ejecuta el código solo cuando es necesario y se escala de manera automática, pasando de pocas solicitudes al día a miles por segundo. Solo paga por el tiempo de computación que consume. No se aplican cargos cuando su código no se está ejecutando. Con AWS Lambdaél, puede ejecutar código para prácticamente cualquier tipo de aplicación o servicio de back-end, todo sin necesidad de administración. Ejecuta el código en una infraestructura informática de alta disponibilidad y realiza todas las tareas de administración de los recursos informáticos, incluido el mantenimiento del servidor y del sistema operativo, el aprovisionamiento de capacidad y el escalado automático, así como la monitorización del código y las funciones de registro. Para obtener más información, consulte [Uso AWS Lambda con Amazon Kinesis](/lambda/latest/dg/with-kinesis.html). Para obtener información sobre la resolución de problemas, consulte [¿Por qué el desencadenador de Kinesis Data Streams no puede invocar mi función de Lambda?](https://aws.amazon.com/premiumsupport/knowledge-center/kinesis-data-streams-lambda-invocation/) # Desarrollar consumidores mediante Amazon Managed Service para Apache Flink Puede utilizar una aplicación de Amazon Managed Service para Apache Flink a fin de procesar y analizar los datos de un flujo de Kinesis mediante SQL, Java o Scala. Las aplicaciones de Managed Service para Apache Flink pueden enriquecer los datos mediante orígenes de referencia, agregar datos a lo largo del tiempo o utilizar el machine learning para encontrar anomalías en los datos. A continuación, puede escribir los resultados del análisis en otro flujo de Kinesis, un flujo de entrega de Firehose o una función de Lambda. Para obtener más información, consulte la [Guía para desarrolladores de Managed Service para Apache Flink para aplicaciones de SQL](/kinesisanalytics/latest/dev/what-is.html) o la [Guía para desarrolladores de Managed Service para Apache Flink para aplicaciones de Flink](/kinesisanalytics/latest/java/what-is.html). # Desarrollar consumidores mediante Amazon Data Firehose Puede utilizar Firehose para leer y procesar los registros de un flujo de Kinesis. Firehose es un servicio totalmente gestionado para entregar datos de streaming en tiempo real a destinos como Amazon S3, Amazon Redshift, OpenSearch Amazon Service y Splunk. Firehose admite cualquier punto de conexión HTTP personalizado o puntos de conexión HTTP que sean propiedad de proveedores de servicios de terceros compatibles, como Datadog, MongoDB y New Relic. También puede configurar Firehose para transformar sus registros de datos y para convertir el formato de los registros antes de distribuir los datos a su destino. Para obtener más información, consulte [Escritura en Firehose mediante Kinesis Data Streams](/firehose/latest/dev/writing-with-kinesis-streams.html). # Lea datos de Kinesis Data Streams mediante otros servicios AWS Los siguientes AWS servicios se pueden integrar directamente con Amazon Kinesis Data Streams para leer los datos de las transmisiones de datos de Kinesis. Revise la información de cada servicio que le interese y consulte las referencias proporcionadas. **Topics** + [ # Leer datos de Kinesis Data Streams con Amazon EMR ](using-other-services-emr.md) + [ # Lea datos de Kinesis Data Streams con Amazon Pipes EventBridge ](using-other-services-ev-pipes.md) + [ # Lea los datos de Kinesis Data Streams mediante AWS Glue ](using-other-services-glue.md) + [ # Leer datos de Kinesis Data Streams con Amazon Redshift ](using-other-services-redshift.md) # Leer datos de Kinesis Data Streams con Amazon EMR Los clústeres de Amazon EMR pueden leer y procesar las transmisiones de Kinesis directamente, mediante herramientas conocidas del ecosistema de Hadoop, como Hive, Pig MapReduce, la API de streaming de Hadoop y Cascading. También puede unir los datos en tiempo real de Kinesis Data Stream con los datos existentes en Amazon S3, Amazon DynamoDB y HDFS en un clúster en ejecución. Puede cargar datos directamente de Amazon EMR en Amazon S3 o en DynamoDB para actividades posteriores al procesamiento. Para obtener más información, consulte [Amazon Kinesis](https://docs.aws.amazon.com/emr/latest/ReleaseGuide/emr-kinesis.html) en la *Guía de publicación de Amazon EMR*. # Lea datos de Kinesis Data Streams con Amazon Pipes EventBridge Amazon EventBridge Pipes admite las transmisiones de datos de Kinesis como fuente. Amazon EventBridge Pipes te ayuda a crear point-to-point integraciones entre los productores de eventos y los consumidores con pasos opcionales de transformación, filtrado y enriquecimiento. Puede usar EventBridge Pipes para recibir registros en una transmisión de datos de Kinesis y, si lo desea, filtrar o mejorar estos registros antes de enviarlos a uno de los destinos disponibles para su procesamiento, incluido Kinesis Data Streams. Para obtener más información, consulte [Amazon Kinesis stream como fuente](https://docs.aws.amazon.com/eventbridge/latest/userguide/eb-pipes-kinesis.html) en la *Amazon EventBridge Release* Guide. # Lea los datos de Kinesis Data Streams mediante AWS Glue Con la AWS Glue transmisión de ETL, puede crear trabajos de extracción, transformación y carga (ETL) en streaming que se ejecuten de forma continua y consuman datos de Amazon Kinesis Data Streams. Los trabajos limpian y transforman los datos y, a continuación, cargan los resultados en los lagos de datos de Amazon S3 o en los almacenes de datos JDBC. Para obtener más información, consulte [Streaming de trabajos ETL en AWS Glue](https://docs.aws.amazon.com/glue/latest/dg/add-job-streaming.html) en la *Guía de versiones de AWS Glue *. # Leer datos de Kinesis Data Streams con Amazon Redshift Amazon Redshift admite la ingesta de streaming desde Amazon Kinesis Data Streams. La característica de ingesta de streaming de Amazon Redshift proporciona una ingesta de datos de streaming de baja latencia y alta velocidad desde Amazon Kinesis Data Streams a una vista materializada de Amazon Redshift. La ingesta de streaming de Amazon Redshift elimina la necesidad de organizar los datos en Amazon S3 antes de ingerirlos en Amazon Redshift. Para obtener más información, consulte [Ingesta de streaming](https://docs.aws.amazon.com/redshift/latest/dg/materialized-view-streaming-ingestion.html) en la *Guía de versiones de Amazon Redshift*. # Leer los datos de Kinesis Data Streams mediante integraciones de terceros Puede leer los flujos de datos de Amazon Kinesis Data Streams 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-read-flink.md) + [ # Adobe Experience Platform ](using-other-services-read-adobe.md) + [ # Apache Druid ](using-other-services-read-druid.md) + [ # Apache Spark ](using-other-services-read-spark.md) + [ # Databricks ](using-other-services-read-databricks.md) + [ # Kafka Confluent Platform ](using-other-services-read-kafka.md) + [ # Kinesumer ](using-other-services-read-kinesumer.md) + [ # Talend ](using-other-services-read-talend.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 el consumo de Kinesis Data Streams mediante Apache Flink, consulte [Amazon Kinesis Data Streams Connector](https://nightlies.apache.org/flink/flink-docs-release-1.15/docs/connectors/datastream/kinesis/). # Adobe Experience Platform 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 el consumo de Kinesis Data Streams mediante Adobe Experience Platform, consulte [Conector de Amazon Kinesis](https://experienceleague.adobe.com/docs/experience-platform/sources/connectors/cloud-storage/kinesis.html). # Apache Druid Druid es una base de datos de análisis en tiempo real de alto rendimiento que permite hacer consultas en fracciones de segundo a datos en streaming y por lotes a escala y bajo carga. Para obtener más información sobre la ingesta de Kinesis Data Streams mediante Apache Druid, consulte [Ingesta de Amazon Kinesis](https://druid.apache.org/docs/latest/development/extensions-core/kinesis-ingestion.html). # Apache Spark Apache Spark es un motor de análisis unificado para el procesamiento de datos a gran escala. Proporciona un alto nivel APIs de Java, Scala, Python y R, y un motor optimizado que admite gráficos de ejecución general. Puede usar Apache Spark para crear aplicaciones de procesamiento de flujos que consuman los datos de sus flujos de datos de Kinesis. Para consumir flujos de datos de Kinesis mediante Apache Spark Structured Streaming, utilice el [conector](https://github.com/awslabs/spark-sql-kinesis-connector) Amazon Kinesis Data Streams. Este conector admite el consumo con la distribución ramificada mejorada, que proporciona a su aplicación un rendimiento de lectura dedicado de hasta 2 MB de datos por segundo por partición. Para más información, consulte [Desarrollo y uso de consumidores personalizados con rendimiento dedicado (Distribución ramificada mejorada)](https://docs.aws.amazon.com/streams/latest/dev/enhanced-consumers.html). Para obtener más información sobre el consumo de Kinesis Data Streams mediante Spark Streaming, consulte [Integración de Spark Streaming \$1 Kinesis](https://spark.apache.org/docs/latest/streaming-kinesis-integration.html). # Databricks Databricks es una plataforma basada en la nube que proporciona un entorno colaborativo para la ingeniería de datos, la ciencia de datos y el machine learning. Para obtener más información sobre el consumo de Kinesis Data Streams con Databricks, consulte [Connect con Amazon Kinesis](https://docs.databricks.com/structured-streaming/kinesis.html). # Kafka Confluent Platform Confluent Platform se basa en Kafka y ofrece funciones y características adicionales que ayudan a las empresas a crear y administrar flujos de datos y aplicaciones de streaming en tiempo real. Para obtener más información sobre el consumo de Kinesis Data Streams mediante Confluent Platform, consulte [Conector de origen de Amazon Kinesis para Confluent Platform](https://docs.confluent.io/kafka-connectors/kinesis/current/overview.html#features). # Kinesumer Kinesumer es un cliente de Go que implementa un cliente de grupo de consumidores distribuido del cliente para Kinesis Data Streams. Para obtener más información, consulte el [repositorio de GitHub de Kinesumer](https://github.com/daangn/kinesumer). # Talend Talend es un software de integración y administración de datos que permite a los usuarios recopilar, transformar y conectar datos de diversos orígenes de forma escalable y eficiente. Para obtener más información sobre el consumo de Kinesis Data Streams con Talend, consulte [Conectar Talend a un Amazon Kinesis Stream](https://help.talend.com/r/en-US/Cloud/connectors-guide/connector-kinesis). # Solución de problemas de los consumidores de Kinesis Data Streams **Topics** + [ ## Error de compilación con el LeaseManagementConfig constructor ](#compilation-error-leasemanagementconfig) + [ ## Algunos registros de Kinesis Data Streams se omiten al usar Kinesis Client Library ](#records-skipped) + [ ## Registros que pertenecen a la misma partición se procesan en distintos procesadores de registros a la vez ](#records-belonging-to-the-same-shard) + [ ## La aplicación consumidora lee a una velocidad menor que lo esperado ](#consumer-app-reading-slower) + [ ## GetRecords devuelve una matriz de registros vacía incluso cuando hay datos en la transmisión ](#getrecords-returns-empty) + [ ## El iterador de partición caduca de forma inesperada ](#shard-iterator-expires-unexpectedly) + [ ## El procesamiento de registros del consumidor se queda atrás ](#record-processing-falls-behind) + [ ## Error de permisos no autorizados para la clave de KMS ](#unauthorized-kms-consumer) + [ ## DynamoDbException: La ruta del documento proporcionada en la expresión de actualización no es válida para la actualización ](#dynamo-db-exception) + [ ## Solucionar otros problemas comunes para los consumidores ](#misc-troubleshooting-consumer) ## Error de compilación con el LeaseManagementConfig constructor Existe la posibilidad de que se produzca un error de compilación relacionado con el constructor de `LeaseManagementConfig` al actualizar Kinesis Client Library (KCL) a la versión 3.x o posterior. Si está creando directamente un objeto de `LeaseManagementConfig` para establecer las configuraciones en lugar de usar `ConfigsBuilder` en la versión 3.x o posterior de KCL, es probable que aparezca el siguiente mensaje de error al compilar el código de la aplicación de KCL. ``` Cannot resolve constructor 'LeaseManagementConfig(String, DynamoDbAsyncClient, KinesisAsyncClient, String)' ``` A partir de la versión 3.x, KCL requiere que se agregue un parámetro más, applicationName (tipo: cadena), después del parámetro tableName. + *Antes*: leaseManagementConfig = new LeaseManagementConfig (TableName, DBClient dynamo, KinesisClient, StreamName, WorkerIdentifier) + *Después de*: leaseManagementConfig = new LeaseManagementConfig (TableName**, ApplicationName, dynamo, KinesisClient**, StreamNameDBClient, WorkerIdentifier) En lugar de crear un LeaseManagementConfig objeto directamente, se recomienda utilizarlo para establecer las configuraciones en KCL 3.x y versiones posteriores. `ConfigsBuilder` `ConfigsBuilder`proporciona una forma más flexible y fácil de mantener de configurar su aplicación KCL. A continuación, se muestra un ejemplo de uso de `ConfigsBuilder` para establecer configuraciones de KCL. ``` ConfigsBuilder configsBuilder = new ConfigsBuilder( streamName, applicationName, kinesisClient, dynamoClient, cloudWatchClient, UUID.randomUUID().toString(), new SampleRecordProcessorFactory() ); Scheduler scheduler = new Scheduler( configsBuilder.checkpointConfig(), configsBuilder.coordinatorConfig(), configsBuilder.leaseManagementConfig() .failoverTimeMillis(60000), // this is an example configsBuilder.lifecycleConfig(), configsBuilder.metricsConfig(), configsBuilder.processorConfig(), configsBuilder.retrievalConfig() ); ``` ## Algunos registros de Kinesis Data Streams se omiten al usar Kinesis Client Library La causa más frecuente de la omisión de registros es que haya una excepción de `processRecords` no administrada. Kinesis Client Library (KCL) depende del código `processRecords` para administrar cualquier excepción que surja del procesamiento de los registros de datos. Cualquier excepción que se origine en `processRecords` se absorbe en KCL. Para evitar los reintentos infinitos sobre un error recurrente, KCL no reenvía el lote de registros procesados en el momento de la excepción. A continuación, KCL llama a `processRecords` para solicitar el siguiente lote de registros de datos sin reiniciar el procesador de registros. Esto da como resultado que en las aplicaciones consumidoras se observen registros omitidos. Para evitar que se omitan registros, administre todas las excepciones de `processRecords` convenientemente. ## Registros que pertenecen a la misma partición se procesan en distintos procesadores de registros a la vez Para cualquier aplicación de Kinesis Client Library (KCL) en ejecución, una partición solo tiene un propietario. Sin embargo, varios procesadores de registros podrían procesar el mismo fragmento temporalmente. Si una instancia de un proceso de trabajo pierde la conectividad de red, KCL asume que el proceso de trabajo inaccesible ya no está procesando registros después de que pase el tiempo de conmutación por error, y da instrucciones a otras instancias de procesos de trabajo para sustituirla. Durante un breve periodo, los nuevos procesadores de registros y los procesadores de registros del proceso de trabajo inaccesible pueden procesar datos procedentes del mismo fragmento. Defina un tiempo de conmutación por error que sea adecuado para su aplicación. En el caso de las aplicaciones de baja latencia, el valor predeterminado de 10 segundos puede representar el tiempo máximo que desee esperar. Sin embargo, en aquellos casos en los que prevea que se producirán problemas de conectividad, como al hacer llamadas en zonas geográficas en las que la conectividad se podría perder con más frecuencia, puede que este número sea demasiado bajo. Su aplicación debe anticiparse a esta situación y administrarla, especialmente debido a que la conectividad de red normalmente se restaura al proceso de trabajo previamente inaccesible. Si los fragmentos de un procesador de registros son ocupados por otro procesador de registros, debe afrontar los dos casos siguientes para cerrarse sin ocasionar problemas: 1. Después de que se complete la llamada actual a `processRecords`, KCL invoca el método de cierre en el procesador de registros con el motivo "ZOMBIE". Cabe esperar que sus procesadores de registros eliminen los recursos según corresponda y, a continuación, se cierren. 1. Si intenta crear un punto de comprobación en un proceso de trabajo "zombie", KCL producirá una `ShutdownException`. Tras recibir esta excepción, lo normal es que el código salga del método actual sin ocasionar problemas. Para obtener más información, consulte [Administrar registros duplicados](kinesis-record-processor-duplicates.md). ## La aplicación consumidora lee a una velocidad menor que lo esperado Los motivos más comunes para que el rendimiento de lectura sea menor que lo esperado son los siguientes: 1. Varias aplicaciones consumidoras tienen lecturas totales que superan los límites por fragmento. Para obtener más información, consulte [Cuotas y límites](service-sizes-and-limits.md). En este caso, puede aumentar el número de particiones en el flujo de datos de Kinesis. 1. El [límite](https://docs.aws.amazon.com/kinesis/latest/APIReference/API_GetRecords.html#API_GetRecords_RequestSyntax) que especifica el número máximo de comandos **GetRecords** por llamada puede haberse configurado con un valor bajo. Si utiliza KCL, es posible que haya configurado el proceso de trabajo con un valor bajo para la propiedad `maxRecords`. En general, recomendamos que utilice los valores predeterminados del sistema para esta propiedad. 1. La lógica de la `processRecords` llamada puede tardar más de lo esperado por varias razones posibles: la lógica puede requerir un uso intensivo de la CPU, I/O bloquear o dificultar la sincronización. Para probar si alguno de estos supuestos es cierto, realice ejecuciones de prueba de procesadores de registros vacíos y compare el rendimiento de lectura. Para obtener información sobre cómo mantener la entrada de datos, consulte [Utilizar la nueva partición, el escalado y el procesamiento paralelo para cambiar el número de particiones](kinesis-record-processor-scaling.md). Si tiene una única aplicación consumidora, siempre puede leer al menos dos veces más rápido que la velocidad de inclusión. Esto se debe a que puede escribir hasta 1000 registros por segundo para escrituras, hasta un máximo de escritura de datos de 1 MB por segundo (incluidas las claves de partición). Cada partición abierta admite hasta 5 transacciones por segundo en el caso de las lecturas, con una velocidad máxima total de lectura de datos de 2 MB por segundo. Tenga en cuenta que con cada lectura (llamada a **GetRecords**) se obtiene un lote de registros. El tamaño de los datos devueltos por **GetRecords** varía en función del uso del fragmento. El volumen máximo de datos que **GetRecords** puede devolver es de 10 MB. Si una llamada devuelve ese límite, las llamadas posteriores realizadas en los siguientes 5 segundos provocan una `ProvisionedThroughputExceededException`. ## GetRecords devuelve una matriz de registros vacía incluso cuando hay datos en la transmisión El consumo o la obtención de registros se basa en un modelo de extracción. Se espera que los desarrolladores [GetRecords](https://docs.aws.amazon.com/kinesis/latest/APIReference/API_GetRecords.html)realicen llamadas en un bucle continuo sin interrupciones. Cada llamada a **GetRecords** devuelve también un valor `ShardIterator` que debe utilizarse en la siguiente iteración del bucle. La operación **GetRecords** no se bloquea. En su lugar, se devuelve de inmediato, con registros de datos relevantes o con un elemento `Records` vacío. Un elemento `Records` vacío se devuelve con dos condiciones: 1. No hay más datos actualmente en el fragmento. 1. No hay datos cerca de la parte del fragmento a la que apunta el `ShardIterator`. La última condición es sutil, pero supone un equilibrio de diseño necesario para evitar el tiempo de búsqueda ilimitado (latencia) al recuperar registros. Por lo tanto, la aplicación que consume la secuencia debe proceder en bucle y llamar a **GetRecords**, ocupándose también de los registros vacíos. En un escenario de producción, la única vez que se debería salir del bucle continuo es cuando el valor `NextShardIterator` es `NULL`. Cuando `NextShardIterator` es `NULL`, significa que el fragmento actual se ha cerrado y el valor `ShardIterator`, de lo contrario, apuntaría más allá del último registro. Si la aplicación consumidora nunca llama a **SplitShard** o a **MergeShards**, el fragmento permanece abierto y las llamadas a **GetRecords** nunca devuelven para `NextShardIterator` el valor `NULL`. Si utiliza Kinesis Client Library (KCL), el patrón de consumo anterior se abstrae automáticamente. Esto incluye la administración automática de un conjunto de fragmentos que cambian de forma dinámica. Con KCL, el desarrollador solo suministra la lógica para procesar registros de entrada. Esto es posible porque la biblioteca realiza automáticamente llamadas continuas a **GetRecords**. ## El iterador de partición caduca de forma inesperada Con cada nueva solicitud **GetRecords** se devuelve un nuevo iterador de fragmentos (como `NextShardIterator`) que debe usar entonces en la siguiente solicitud **GetRecords** (como `ShardIterator`). Normalmente, este iterador de fragmentos no caduca antes de utilizarlo. Sin embargo, los iteradores de fragmentos pueden caducar por no haber llamado a **GetRecords** durante más de 5 minutos, o porque haber reiniciado la aplicación consumidora. Si el iterador de particiones caduca inmediatamente antes de poder utilizarlo, esto podría indicar que la tabla de DynamoDB que utiliza Kinesis no tiene capacidad suficiente para almacenar los datos arrendados. Es más probable que se dé esta situación si tiene un gran número de fragmentos. Para solucionar este problema, aumente la capacidad de escritura asignada a la tabla de fragmentos. Para obtener más información, consulte [Usar una tabla de arrendamiento para realizar el seguimiento de las particiones procesadas por la aplicación de consumo de KCL](shared-throughput-kcl-consumers.md#shared-throughput-kcl-consumers-leasetable). ## El procesamiento de registros del consumidor se queda atrás En la mayoría de casos de uso, las aplicaciones consumidoras leen los datos más recientes de la secuencia. En determinadas circunstancias, puede que las lecturas del consumidor se queden atrás, lo que no es deseable. Tras identificar el retraso con el que están realizando las lecturas sus consumidores, consulte los motivos más comunes por los que estos se retrasan. Comience por la métrica `GetRecords.IteratorAgeMilliseconds`, que controla la posición de lectura de todos los fragmentos y los consumidores de la secuencia. Tenga en cuenta que si la antigüedad de un iterador supera el 50 % del periodo de retención (con un valor predeterminado de 24 horas pero configurable hasta 365 días), existe el riesgo de pérdida de datos debido a la caducidad del registro. Una parche rápido es aumentar el periodo de retención. Así se detiene la pérdida de datos importantes mientras se realizan los pasos para solucionar el problema. Para obtener más información, consulte [Supervise el servicio Amazon Kinesis Data Streams con Amazon CloudWatch](monitoring-with-cloudwatch.md). A continuación, identifique el retraso con el que su aplicación de consumo lee cada fragmento mediante una CloudWatch métrica personalizada emitida por la Biblioteca de clientes de Kinesis (KCL). `MillisBehindLatest` Para obtener más información, consulte [Supervise la biblioteca de clientes de Kinesis con Amazon CloudWatch](monitoring-with-kcl.md). Estos son los motivos más comunes por los que los consumidores se pueden retrasar: + Los grandes aumentos repentinos de `GetRecords.IteratorAgeMilliseconds` o `MillisBehindLatest` suelen indicar un problema transitorio, como que la operación de la API provoca un error en una aplicación descendente. Investigue estos aumentos repentinos si alguna de las métricas muestra constantemente este comportamiento. + Un incremento gradual de estas métricas indica que un consumidor no mantiene el ritmo de la secuencia porque no procesa los registros lo suficientemente rápido. Las causas más comunes para este comportamiento son la insuficiencia de recursos físicos o una lógica de procesamiento de registros que no está ajustada a un aumento en el rendimiento de la secuencia. Para comprobar este comportamiento, consulte las demás CloudWatch métricas personalizadas que la KCL emite asociadas a la `processTask` operación, incluidas, y. `RecordProcessor.processRecords.Time` `Success` `RecordsProcessed` + Si percibe un aumento en la métrica `processRecords.Time` que se correlaciona con una mejora en el rendimiento, debe analizar su lógica de procesamiento de registros para identificar por qué no se ajusta al aumento de rendimiento. + Si percibe un incremento de los valores `processRecords.Time` que no está correlacionado con el aumento de rendimiento, compruebe si está realizando llamadas de bloqueo en la ruta crítica, ya que suelen ser la causa de la reducción de velocidad en el procesamiento de registros. Otro enfoque consiste en aumentar el paralelismo incrementando el número de fragmentos. Por último, confirme que tiene una cantidad suficiente de recursos físicos (memoria, uso de CPU, entre otros) en los nodos de procesamiento subyacentes durante los picos de demanda. ## Error de permisos no autorizados para la clave de KMS Este error se produce cuando una aplicación de consumo lee una transmisión cifrada sin permisos sobre la AWS KMS clave. 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). ## DynamoDbException: La ruta del documento proporcionada en la expresión de actualización no es válida para la actualización Al utilizar KCL 3.x con AWS SDK para Java las versiones 2.27.19 a 2.27.23, puede producirse la siguiente excepción de DynamoDB: «software.amazon.awssdk.services.dynamodb.model. DynamoDbException: La ruta del documento proporcionada en la expresión de actualización no es válida para la actualización (servicio:, código de estado: 400, identificador de solicitud: xxx)» DynamoDb Este error se produce debido a un problema conocido AWS SDK para Java que afecta a la tabla de metadatos de DynamoDB administrada por KCL 3.x. El problema se produjo en la versión 2.27.19 y afecta a todas las versiones hasta la 2.27.23. El problema se resolvió en la versión 2.27.24 de AWS SDK para Java . Para obtener un rendimiento y una estabilidad óptimos, se recomienda actualizar a la versión 2.28.0 o posterior. ## Solucionar otros problemas comunes para los consumidores + [¿Por qué el desencadenador de Kinesis Data Streams no puede invocar mi función de Lambda?](https://aws.amazon.com/premiumsupport/knowledge-center/kinesis-data-streams-lambda-invocation/) + [¿Cómo puedo detectar y solucionar las ReadProvisionedThroughputExceeded excepciones en Kinesis Data Streams?](https://aws.amazon.com/premiumsupport/knowledge-center/kinesis-readprovisionedthroughputexceeded/) + [¿Por qué tengo problemas de alta latencia con Kinesis Data Streams?](https://aws.amazon.com/premiumsupport/knowledge-center/kinesis-data-stream-latency-issues/) + [¿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 puedo solucionar los problemas de una aplicación de KCL bloqueada para Kinesis Data Streams?](https://aws.amazon.com/premiumsupport/knowledge-center/kcl-kinesis-data-streams/) + [¿Puedo usar diferentes aplicaciones de Amazon Kinesis Client Library con la misma tabla de Amazon DynamoDB?](https://aws.amazon.com/premiumsupport/knowledge-center/kinesis-kcl-apps-dynamodb-table/) # Optimizar consumidores de Amazon Kinesis Data Streams. Se puede optimizar incluso un poco más su consumidor de Amazon Kinesis Data Streams según el comportamiento específico que ve. Revise los siguientes temas para identificar soluciones. **Topics** + [ # Mejorar el procesamiento de baja latencia ](kinesis-low-latency.md) + [ # Procese datos serializados AWS Lambda con la biblioteca Amazon Kinesis Producer ](kinesis-record-deaggregation.md) + [ # Utilizar la nueva partición, el escalado y el procesamiento paralelo para cambiar el número de particiones ](kinesis-record-processor-scaling.md) + [ # Administrar registros duplicados ](kinesis-record-processor-duplicates.md) + [ # Administrar el startup, el shutdown y la limitación ](kinesis-record-processor-additional-considerations.md) # Mejorar el procesamiento de baja latencia El *retraso de propagación* se define como la end-to-end latencia desde el momento en que se escribe un registro en la transmisión hasta que lo lee una aplicación de consumo. Este retraso varía en función de una serie de factores, pero principalmente se ve afectado por el intervalo de sondeo de las aplicaciones consumidoras. Para la mayoría de las aplicaciones, le recomendamos que se sondee cada fragmento una vez por segundo y aplicación. Esto le permite tener varias aplicaciones de consumo que procesen un flujo simultáneamente sin alcanzar los límites de Amazon Kinesis Data Streams de cinco llamadas `GetRecords` por segundo. Además, el procesamiento de lotes grandes de datos suele ser más eficaz en la reducción de la latencia de red y otras latencias descendentes en su aplicación. Los valores predeterminados en KCL se establecen para seguir la práctica recomendada de sondear cada segundo. Este valor predeterminado tiene como consecuencia retrasos promedios en la propagación que normalmente están por debajo de 1 segundo. Los registros de Kinesis Data Streams están disponibles para su lectura inmediatamente después de que se escriben. Existen algunos casos de uso en los que es necesario aprovecharse de este factor y requieren que se consuman los datos de la secuencia en cuanto estén disponibles. Puede reducir significativamente el retraso de propagación si anula la configuración predeterminada de KCL para sondear con mayor frecuencia, tal y como se muestra en los siguientes ejemplos. Código de configuración de KCL en Java: ``` kinesisClientLibConfiguration = new KinesisClientLibConfiguration(applicationName, streamName, credentialsProvider, workerId).withInitialPositionInStream(initialPositionInStream).withIdleTimeBetweenReadsInMillis(250); ``` Configuración del archivo de propiedades para KCL en Python y Ruby: ``` idleTimeBetweenReadsInMillis = 250 ``` **nota** Dado que Kinesis Data Streams tiene un límite de cinco llamadas a `GetRecords` por segundo y por partición, si se establece en la propiedad `idleTimeBetweenReadsInMillis` un valor inferior a 200 ms puede ocurrir que la aplicación encuentre la excepción `ProvisionedThroughputExceededException`. Si se producen demasiadas de estas excepciones, pueden darse retrocesos exponenciales y, por lo tanto, causar latencias importantes e inesperadas en el procesamiento. Si configura esta propiedad para que sea de 200 ms o esté por encima de este valor y tiene más de una aplicación de procesamiento, experimentará una limitación controlada similar. # Procese datos serializados AWS Lambda con la biblioteca Amazon Kinesis Producer [Amazon Kinesis Producer Library](https://docs.aws.amazon.com/kinesis/latest/dev/developing-producers-with-kpl.html) (KPL) agrega pequeños registros formateados por el usuario en registros de mayor tamaño, hasta 1 MB, para utilizar mejor el rendimiento de Amazon Kinesis Data Streams. Si bien el KCL para Java admite la desagregación de estos registros, necesitará usar un módulo especial para desagregar los registros cuando lo utilice AWS Lambda como consumidor de sus transmisiones. Puede obtener el código de proyecto y las instrucciones necesarios GitHub en los módulos de [desagregación para Lambda de la biblioteca Amazon Kinesis Producer](https://github.com/awslabs/kinesis-deaggregation). AWS Los componentes de este proyecto le permiten procesar datos serializados de KPL en Java AWS Lambda, Node.js y Python. Estos componentes también se pueden utilizar como parte de una [aplicación KCL multilenguaje](https://github.com/awslabs/amazon-kinesis-client/blob/master/amazon-kinesis-client-multilang/src/main/java/software/amazon/kinesis/multilang/package-info.java). # Utilizar la nueva partición, el escalado y el procesamiento paralelo para cambiar el número de particiones Los *cambios en los fragmentos* le permiten aumentar o reducir el número de fragmentos en una secuencia para adaptarse a los cambios en la velocidad del flujo de datos en la secuencia. Los cambios en los fragmentos los suele realizar una aplicación administrativa que monitorea las métricas de administración de datos en los fragmentos. Aunque la propia KCL no inicia las operaciones de cambios en las particiones, está diseñada para adaptarse a los cambios en el número de particiones que resultan de las mismas. Como se indica en [Usar una tabla de arrendamiento para realizar el seguimiento de las particiones procesadas por la aplicación de consumo de KCL](shared-throughput-kcl-consumers.md#shared-throughput-kcl-consumers-leasetable), KCL hace un seguimiento de las particiones en el flujo mediante una tabla de Amazon DynamoDB. Cuando se crean nuevas particiones como resultado de estos cambios, KCL descubre las nuevas particiones y rellena nuevas filas en la tabla. Los procesos de trabajo descubren automáticamente los nuevos fragmentos y crean procesadores para administrar los datos de los mismos. KCL también distribuye las particiones en el flujo, por todos los procesos de trabajo y procesadores de registros disponibles. KCL garantiza que los datos existentes en particiones antes de que se produjeran cambios en los mismos se procesen primero. Después de que los datos se hayan procesado, los datos de los nuevos fragmentos se envían a los procesadores de registros. De esta forma, KCL conserva el orden en que los registros de datos se han agregado al flujo para una clave de partición en particular. ## Ejemplo: Cambio en las particiones, escalado y procesamiento paralelo El siguiente ejemplo ilustra cómo KCL ayuda a administrar el escalado y los cambios en las particiones: + Por ejemplo, si su aplicación se ejecuta en una instancia de EC2 y está procesando un flujo de datos de Kinesis que tiene cuatro particiones. Esta instancia tiene un proceso de trabajo de KCL y cuatro procesadores de registros (un procesador de registros por partición). Estos cuatro procesadores de registros se ejecutan en paralelo en el mismo proceso. + A continuación, si escala la aplicación para utilizar otra instancia, tendrá dos instancias procesando una secuencia con cuatro fragmentos. Cuando el proceso de trabajo de KCL se inicia en la segunda instancia, realiza un equilibrio de la carga con la primera instancia, de modo que cada instancia ahora procesará dos particiones. + Si entonces decide dividir los cuatro fragmentos en cinco fragmentos. La nueva KCL vuelve a coordinar el procesamiento en las instancias: una instancia procesa tres particiones y la otra procesa dos particiones. Se produce una coordinación similar al fusionar fragmentos. Normalmente, cuando utilice KCL, debe asegurarse de que el número de instancias no supere el número de particiones (excepto si quiere esperar a errores). Cada partición se procesa exactamente en un proceso de trabajo de KCL y tiene un procesador de registros correspondientes, por lo que no necesitará varias instancias para procesar una partición. Sin embargo, un proceso de trabajo puede procesar cualquier número de fragmentos, por lo que no hay problema si el número de fragmentos supera el número de instancias. Para aumentar el procesamiento de su aplicación, debe probar una combinación de estos enfoques: + Aumentar el tamaño de la instancia (porque todos los procesadores de registros se ejecutan en paralelo dentro de un proceso) + Al aumentar el número de instancias hasta el número máximo de fragmentos abiertos (porque los fragmentos se pueden procesar de manera independiente) + Al aumentar el número de fragmentos (lo que aumenta el nivel de paralelismo) Tenga en cuenta que puede utilizar Auto Scaling para escalar automáticamente las instancias en función de las métricas adecuadas. Para obtener más información, consulte la [Guía del usuario de Amazon EC2 Auto Scaling](https://docs.aws.amazon.com/autoscaling/ec2/userguide/). Cuando los cambios en los fragmentos aumentan el número de fragmentos en la secuencia, el aumento correspondiente en el número de procesadores de registros aumenta la carga de las instancias EC2 que los alojan. Si las instancias forman parte de un grupo de Auto Scaling y la carga aumenta lo suficiente, el grupo de Auto Scaling añade más instancias para administrar el aumento de la carga. Debe configurar las instancias para lanzar su aplicación de Amazon Kinesis Data Streams en el inicio, de forma que los procesos de trabajo y los procesadores de registros adicionales se activen en la nueva instancia de inmediato. Para obtener más información acerca de los cambios en los fragmentos, consulte [Cambiar las particiones de un flujo](kinesis-using-sdk-java-resharding.md). # Administrar registros duplicados Hay dos principales motivos por los que se podrían entregar registros más de una vez en la aplicación de Amazon Kinesis Data Streams: los reintentos de los productores y de los consumidores. Su aplicación debe prever y administrar de forma adecuada el procesamiento de registros individuales varias veces. ## Reintentos en productores Piense en un productor que experimenta un tiempo de inactividad relacionado con un problema en la red después de hacer una llamada a `PutRecord`, pero antes de que pueda recibir la confirmación desde Amazon Kinesis Data Streams. El productor no puede asegurar que se ha entregado el registro a Kinesis Data Streams. Suponiendo que cada registro es importante para la aplicación, el productor habría sido escrito para volver a intentar la llamada con los mismos datos. Si ambas llamadas a `PutRecord` sobre los mismos datos se enviaron correctamente a Kinesis Data Streams, habrá dos registros de Kinesis Data Streams. Aunque los dos registros tendrán datos idénticos, también tendrán números secuenciales únicos. Las aplicaciones que necesitan garantías estrictas deben tener incrustada una clave maestra en el registro para eliminar los duplicados en etapas posteriores del procesamiento. Tenga en cuenta que el número de duplicados debidos a reintentos del productor suele ser bajo en comparación con el número de duplicados debidos a reintentos del consumidor. **nota** Si utilizas el AWS SDK`PutRecord`, consulta el [comportamiento de los reintentos](https://docs.aws.amazon.com/sdkref/latest/guide/feature-retry-behavior.html) del SDK en la guía del usuario de *AWS SDKs and Tools*. ## Reintentos en consumidores Los reintentos en los consumidores (aplicaciones de procesamiento de datos) ocurren cuando se reinician los procesadores de registros. Los procesadores de registros para un mismo fragmento se reinician en los siguientes casos: 1. Si un proceso de trabajo termina de forma inesperada 1. Si las instancias de los procesos de trabajo se agregan o eliminan 1. Si se fusionan o dividen fragmentos 1. Si se implementa la aplicación En todos estos casos, el mapeo del shards-to-worker-to procesador de registros se actualiza continuamente para equilibrar la carga del procesamiento. Los procesadores de fragmentos que se migraron a otras instancias reinician el procesamiento de registros a partir del último punto de comprobación. Esto se traduce en un procesamiento de registros duplicado, tal y como se muestra en el ejemplo siguiente. Para obtener más información sobre el balanceo de carga, consulte [Utilizar la nueva partición, el escalado y el procesamiento paralelo para cambiar el número de particiones](kinesis-record-processor-scaling.md). ### Ejemplo: Reintentos en consumidores que dan como resultado la entrega de registros En este ejemplo tiene una aplicación que lee continuamente los registros de un flujo, agrega registros en un archivo local y carga el archivo en Amazon S3. Para simplificar, supongamos que solo hay 1 fragmento y un proceso de trabajo que lo procesa. Fíjese en el siguiente ejemplo de secuencia de eventos, suponiendo que el último punto de comprobación se realizó en el número de registro 10 000: 1. Un proceso de trabajo lee el siguiente lote de registros del fragmento, los registros de 10 001 a 20 000. 1. A continuación, el proceso de trabajo transmite el lote de registros al procesador de registros asociado. 1. El procesador de registros agrega los datos, crea un archivo de Amazon S3 y carga el archivo a Amazon S3 correctamente. 1. El proceso de trabajo se cierra de forma inesperada antes de que pueda crearse un nuevo punto de comprobación. 1. La aplicación, el proceso de trabajo y el procesador de registros se reinician. 1. A partir de ahora, el proceso de trabajo comienza a leer desde el último punto de comprobación correcto, en este caso el 10 001. Por lo tanto, los registros 10 001-20 000 se consumen más de una vez. ### Resistencia a los reintentos en consumidores Aunque puede que los registros se procesen más de una vez, la aplicación podría presentar los efectos adversos como si los registros se hubieran procesado solo una vez (procesamiento idempotente). Las soluciones a este problema varían en cuanto a su complejidad y precisión. Si el destino final de los datos puede administrar bien los duplicados, le recomendamos que deje que dicho destino final se encargue del procesamiento idempotente. Por ejemplo, con [Opensearch](https://www.opensearch.org/) puedes usar una combinación de control de versiones y exclusivo IDs para evitar el procesamiento duplicado. En el ejemplo de la aplicación de la sección anterior, lee continuamente los registros de un flujo, agrega los registros en un archivo local y carga el archivo en Amazon S3. Como se ha indicado, los registros 10 001-20 000 se consumen más de una vez, lo que da como resultado que haya varios archivos en Amazon S3 con los mismos datos. Una forma de mitigar los duplicados en este ejemplo consiste en garantizar que en el paso 3 se utilice el siguiente esquema: 1. El procesador de registros utiliza un número fijo de registros por archivo de Amazon S3; por ejemplo, 5000. 1. El nombre de archivo utiliza este esquema: prefijo de Amazon S3, ID de partición y `First-Sequence-Num`. En este caso, podría ser algo parecido a `sample-shard000001-10001`. 1. Después de cargar el archivo de Amazon S3, cree un punto de verificación especificando `Last-Sequence-Num`. En este caso, debería establecer puntos de comprobación en el número de registro 15 000. Con este esquema, incluso si los registros se procesan más de una vez, el archivo de Amazon S3 resultante tendrá el mismo nombre y los mismos datos. Los reintentos solo dan como resultado la escritura de los mismos datos en el mismo archivo más de una vez. En el caso de una operación de cambio en los fragmentos, el número de registros que quedan en el fragmento podría ser menor que el número fijo necesario. En este caso, el método `shutdown()` tiene que volcar el archivo a Amazon S3 y establecer puntos de verificación en el último número secuencial. El esquema anterior también es compatible con las operaciones de cambios en los fragmentos. # Administrar el startup, el shutdown y la limitación A continuación, se muestran otras consideraciones que podría incorporar en el diseño de su aplicación de Amazon Kinesis Data Streams. **Topics** + [ ## Productores de datos de startup y consumidores de datos ](#kinesis-record-processor-producer-consumer-coordination) + [ ## Cerrar una aplicación de Amazon Kinesis Data Streams ](#developing-consumers-with-kcl-shutdown) + [ ## Limitación de lectura ](#kinesis-record-processor-read-throttling) ## Productores de datos de startup y consumidores de datos De forma predeterminada, KCL comienza a leer registros desde el extremo del flujo, que es el registro agregado más recientemente. Con esta configuración, si una aplicación productora de datos agrega registros a la secuencia antes de que se ejecuten los procesadores de registros receptores, los procesadores de registros no podrán leer los registros tras el inicio. Para cambiar el comportamiento de los procesadores de registros de forma que siempre lean los datos desde el inicio del flujo, defina el siguiente valor en el archivo de propiedades de su aplicación de Amazon Kinesis Data Streams: ``` initialPositionInStream = TRIM_HORIZON ``` De forma predeterminada, Amazon Kinesis Data Streams almacena todos los datos durante 24 horas. También admite la retención prolongada de hasta 7 días y la retención a largo plazo de hasta 365 días. Este periodo de tiempo se denomina el *periodo de retención*. Si se establece la posición inicial en `TRIM_HORIZON`, el procesador de registros se iniciará con los datos más antiguos de la secuencia, según defina el periodo de retención. Incluso con la configuración `TRIM_HORIZON`, si un procesador de registros se iniciara tras transcurrir mucho más tiempo del que determina el periodo de retención, algunos de los registros de la secuencia dejarán de estar disponibles. Por este motivo, siempre debes hacer que las aplicaciones de consumo lean de la transmisión y usar la CloudWatch métrica `GetRecords.IteratorAgeMilliseconds` para controlar que las aplicaciones estén al día con los datos entrantes. En algunos casos, puede ser una buena opción para los procesadores de registros omitir los primeros registros de la secuencia. Por ejemplo, puede ejecutar algunos registros iniciales en la transmisión para comprobar que la transmisión funciona end-to-end según lo esperado. Tras realizar la verificación inicial, a continuación iniciaría sus procesos de trabajo y empezaría a introducir datos de producción en la secuencia. Para obtener más información acerca de la opción `TRIM_HORIZON`, consulte [Usar iteradores de particiones](developing-consumers-with-sdk.md#kinesis-using-sdk-java-get-data-shard-iterators). ## Cerrar una aplicación de Amazon Kinesis Data Streams Cuando su aplicación de Amazon Kinesis Data Streams haya completado la tarea correspondiente, debe cerrar mediante la terminación de las instancias de EC2 en las que se está ejecutando. Puede terminar las instancias usando la [Consola de administración de AWS](https://console.aws.amazon.com//ec2/home) o la [AWS CLI](https://docs.aws.amazon.com/cli/latest/reference/ec2/index.html). Después de cerrar la aplicación de Amazon Kinesis Data Streams, debe eliminar la tabla de Amazon DynamoDB que utilizó KCL para realizar un seguimiento del estado de la aplicación. ## Limitación de lectura El rendimiento de una secuencia se facilita en el nivel del fragmento. Cada partición tiene un rendimiento de lectura de hasta cinco transacciones por segundo, con una velocidad máxima total de lectura de datos de 2 MB por segundo. Si una aplicación (o grupo de aplicaciones que operan en el mismo flujo) intenta obtener datos de una partición a una velocidad superior, Kinesis Data Streams limita las operaciones Get correspondientes. En una aplicación de Amazon Kinesis Data Streams, si un procesador de registros procesa datos más rápido que el límite (por ejemplo, en el caso de una conmutación por error), se produce una limitación. Dado que KCL administra las interacciones entre la aplicación y Kinesis Data Streams, las excepciones por limitación se producen en el código de KCL y no en el código de la aplicación. No obstante, puesto que KCL registra estas excepciones, las verá en los registros. Si cree que su aplicación es objeto de una limitación controlada de forma constante, debería pensar en aumentar el número de fragmentos para la secuencia.