Les traductions sont fournies par des outils de traduction automatique. En cas de conflit entre le contenu d'une traduction et celui de la version originale en anglais, la version anglaise prévaudra.
# Écrire des données dans Amazon Kinesis Data Streams
Un *producteur* est une application qui écrit des données dans Amazon Kinesis Data Streams. Vous pouvez créer des producteurs pour Kinesis Data Streams à l'aide de la Kinesis Producer Library (KPL) et de AWS SDK pour Java la Kinesis Producer Library (KPL).
Si vous ne connaissez pas Kinesis Data Streams, commencez par vous familiariser avec les concepts et la terminologie présentés dans [Qu'est-ce qu'Amazon Kinesis Data Streams ?](introduction.md) et [Utilisez le AWS CLI pour effectuer des opérations Amazon Kinesis Data Streams](getting-started.md).
**Important**
Kinesis Data Streams prend en charge les modifications de la période de conservation des enregistrements de données de votre flux de données. Pour de plus amples informations, veuillez consulter [Modifier la période de conservation des données](kinesis-extended-retention.md).
Pour placer des données dans le flux, vous devez spécifier le nom du flux, une clé de partition et le blob de données à ajouter au flux. La clé de partition sert à déterminer la partition à laquelle l'enregistrement de données est ajouté dans le flux.
Toutes les données de la partition sont envoyées à la même application de travail qui traite la partition. La clé de partition que vous utilisez dépend de votre logique d'application. En général, le nombre de clés de partition doit être bien supérieur au nombre de partitions. Cela provient du fait que la clé de partition sert à déterminer comment mapper un enregistrement de données à une partition spécifique. Si vous avez suffisamment de clés de partition, les données peuvent être réparties uniformément entre les partitions d'un flux.
**Topics**
+ [
# Développez des producteurs à l'aide de la bibliothèque Amazon Kinesis Producer Library (KPL)
](developing-producers-with-kpl.md)
+ [
# Développez des producteurs à l'aide de l'API Amazon Kinesis Data Streams avec le AWS SDK pour Java
](developing-producers-with-sdk.md)
+ [
# Écrire sur Amazon Kinesis Data Streams à l'aide de Kinesis Agent
](writing-with-agents.md)
+ [
# Écrivez sur Kinesis Data Streams à l'aide d'autres services AWS
](using-other-services.md)
+ [
# Écrivez sur Kinesis Data Streams à l'aide d'intégrations tierces
](using-other-services-third-party.md)
+ [
# Résoudre les problèmes liés aux producteurs d'Amazon Kinesis Data Streams
](troubleshooting-producers.md)
+ [
# Optimisez les producteurs de Kinesis Data Streams
](advanced-producers.md)
# Développez des producteurs à l'aide de la bibliothèque Amazon Kinesis Producer Library (KPL)
Un producteur Amazon Kinesis Data Streams est une application qui place des enregistrements de données utilisateur dans un flux de données Kinesis (également appelé *ingestion de données*). L'Amazon Kinesis Producer Library (KPL) simplifie le développement des applications destinées aux producteurs, en permettant aux développeurs d'atteindre un débit d'écriture élevé dans un flux de données Kinesis.
Vous pouvez surveiller le KPL avec Amazon CloudWatch. Pour de plus amples informations, veuillez consulter [Surveillez la bibliothèque Kinesis Producer avec Amazon CloudWatch](monitoring-with-kpl.md).
**Topics**
+ [
## Revoir le rôle du KPL
](#developing-producers-with-kpl-role)
+ [
## Découvrez les avantages de l'utilisation du KPL
](#developing-producers-with-kpl-advantage)
+ [
## Sachez quand ne pas utiliser le KPL
](#developing-producers-with-kpl-when)
+ [
# Installez le KPL
](kinesis-kpl-dl-install.md)
+ [
# Migrer de KPL 0.x vers KPL 1.x
](kpl-migration-1x.md)
+ [
# Transition vers les certificats Amazon Trust Services (ATS) pour le KPL
](kinesis-kpl-upgrades.md)
+ [
# Plateformes prises en charge par KPL
](kinesis-kpl-supported-plats.md)
+ [
# Concepts clés du KPL
](kinesis-kpl-concepts.md)
+ [
# Intégrer le KPL au code du producteur
](kinesis-kpl-integration.md)
+ [
# Écrivez dans votre flux de données Kinesis à l'aide du KPL
](kinesis-kpl-writing.md)
+ [
# Configuration de la bibliothèque Amazon Kinesis Producer
](kinesis-kpl-config.md)
+ [
# Mettre en œuvre la désagrégation des consommateurs
](kinesis-kpl-consumer-deaggregation.md)
+ [
# Utiliser le KPL avec Amazon Data Firehose
](kpl-with-firehose.md)
+ [
# Utiliser le KPL avec le registre des AWS Glue schémas
](kpl-with-schemaregistry.md)
+ [
# Configurer la configuration du proxy KPL
](kpl-proxy-configuration.md)
+ [
# Politique de cycle de vie des versions KPL
](kpl-version-lifecycle-policy.md)
**Note**
Il est recommandé de passer à la dernière version de KPL. La KPL est régulièrement mise à jour avec de nouvelles versions qui incluent les derniers correctifs de dépendance et de sécurité, des corrections de bogues et de nouvelles fonctionnalités rétrocompatibles. Pour plus d'informations, consultez [https://github.com/awslabs/amazon-kinesis-producer/releases/](https://github.com/awslabs/amazon-kinesis-producer/releases/).
## Revoir le rôle du KPL
Le KPL est une easy-to-use bibliothèque hautement configurable qui vous permet d'écrire dans un flux de données Kinesis. Elle sert d'intermédiaire entre le code de votre application producteur et les actions d'API Kinesis Data Streams. La KPL effectue les tâches principales suivantes :
+ Ecrit dans un ou plusieurs flux de données Kinesis avec un mécanisme de nouvelle tentative configurable
+ Collecte les enregistrements et utilise `PutRecords` pour écrire plusieurs enregistrements dans plusieurs partitions par demande
+ Regroupe les enregistrements d'utilisateur pour accroître la taille de la charge utile et améliorer le débit
+ S'intègre de façon transparente avec la [bibliothèque client Kinesis](https://docs.aws.amazon.com/kinesis/latest/dev/developing-consumers-with-kcl.html) (KCL) pour dégrouper les enregistrements en lot sur l'application producteur
+ Soumet CloudWatch les statistiques Amazon en votre nom afin de fournir une visibilité sur les performances des producteurs
Notez que le KPL est différent de l'API Kinesis Data Streams disponible dans le. [AWS SDKs](https://aws.amazon.com/tools/) L'API Kinesis Data Streams vous permet de gérer de nombreux aspects de Kinesis Data Streams (notamment la création de flux, le repartitionnement, le placement et l'extraction des enregistrements), tandis que la KPL fournit une couche d'abstraction spécifiquement pour l'ingestion de données. Pour plus d'informations sur l'API Kinesis Data Streams, consultez la [Référence d'API Amazon Kinesis](https://docs.aws.amazon.com/kinesis/latest/APIReference/) (français non garanti).
## Découvrez les avantages de l'utilisation du KPL
La liste suivante présente certains des principaux avantages liés à l'utilisation de la KPL pour le développement des applications producteur Kinesis Data Streams.
La KPL peut servir aussi bien dans les deux cas d'utilisation synchrones qu'asynchrones. Nous recommandons d'utiliser les meilleures performances de l'interface asynchrone, à moins d'avoir une raison précise d'utiliser le comportement synchrone. Pour plus d'informations sur ces deux cas d'utilisation et l'exemple de code, consultez la page [Écrivez dans votre flux de données Kinesis à l'aide du KPL](kinesis-kpl-writing.md).
**Avantages en termes de performances**
La KPL peut aider à créer des applications producteur hautes performances. Imaginez une situation où vos instances Amazon EC2 servent de proxy pour la collecte des événements de 100 octets sur des centaines ou milliers d'appareils de faible puissance et pour l'écriture des enregistrements dans un flux de données Kinesis. Ces instances EC2 doivent écrire chacune des milliers d'événements par seconde dans votre flux de données. Pour atteindre le débit nécessaire, les applications producteur doivent mettre en œuvre une logique complexe, telle que le traitement par lots ou multithreading, en plus d'une logique de nouvelle tentative et du dégroupement des enregistrements du côté consommateur. La KPL effectue toutes ces tâches à votre place.
**Facilité d'utilisation côté consommateur**
Pour les développeurs côté consommateur qui utilisent la KCL en Java, la KPL s'intègre sans effort supplémentaire. Lorsque la KCL extrait un enregistrement Kinesis Data Streams groupé, composé de plusieurs enregistrements d'utilisateur KPL, elle appelle automatiquement la KPL pour extraire les différents enregistrements d'utilisateur avant de les renvoyer à l'utilisateur.
Pour les développeurs côté consommateur qui n'utilisent pas la KCL, mais directement l'opération d'API `GetRecords`, une bibliothèque KPL Java est disponible pour extraire les différents enregistrements d'utilisateur avant de les renvoyer à l'utilisateur.
**Surveillance d'une application producteur**
Vous pouvez collecter, surveiller et analyser vos producteurs de Kinesis Data Streams à l'aide d' CloudWatch Amazon et du KPL. Le KPL émet des indicateurs de débit, d'erreur et d'autres indicateurs en votre CloudWatch nom, et est configurable pour être surveillé au niveau du flux, de la partition ou du producteur.
**Architecture asynchrone**
Comme le KPL peut mettre en mémoire tampon les enregistrements avant de les envoyer à Kinesis Data Streams, il n'oblige pas l'application appelant à bloquer et à attendre la confirmation que l'enregistrement est arrivé sur le serveur avant de poursuivre l'exécution. Un appel pour placer un enregistrement dans la KPL renvoie toujours immédiatement une réponse et n'attend pas l'envoi de l'enregistrement ou une réponse reçue du serveur. En revanche, un objet `Future` est créé, qui reçoit le résultat de l'envoi ultérieur de l'enregistrement à Kinesis Data Streams. Ce comportement est identique à celui des clients asynchrones dans le AWS SDK.
## Sachez quand ne pas utiliser le KPL
La KPL peut subir un délai de traitement supplémentaire allant jusqu'à `RecordMaxBufferedTime` dans la bibliothèque (configurable par l'utilisateur). Les valeurs élevées de `RecordMaxBufferedTime` se traduisent par une efficacité de compression supérieure et de meilleures performances. Les applications qui ne tolèrent pas ce délai supplémentaire peuvent avoir besoin d'utiliser directement le AWS SDK. Pour plus d'informations sur l'utilisation du AWS SDK avec Kinesis Data Streams, consultez. [Développez des producteurs à l'aide de l'API Amazon Kinesis Data Streams avec le AWS SDK pour Java](developing-producers-with-sdk.md) Pour plus d'informations sur `RecordMaxBufferedTime` et les autres propriétés configurables par l'utilisateur de la KPL, consultez [Configuration de la bibliothèque Amazon Kinesis Producer](kinesis-kpl-config.md).
# Installez le KPL
Amazon fournit des fichiers binaires prédéfinis de la bibliothèque Amazon Kinesis Producer (KPL) en C\$1\$1 pour macOS, Windows et les distributions Linux récentes (pour plus d'informations sur les plateformes prises en charge, consultez la section suivante). Ces fichiers binaires sont packagés sous forme de fichiers .jar Java et sont appelés automatiquement et utilisés si vous utilisez Maven pour installer le package. Pour rechercher les dernières versions de la KPL et de la KCL, utilisez les liens de recherche Maven suivants :
+ [KPL](https://search.maven.org/#search|ga|1|amazon-kinesis-producer)
+ [KCL](https://search.maven.org/#search|ga|1|amazon-kinesis-client)
Les fichiers binaires Linux ont été compilés avec GCC (GNU Compiler Collection) et liés statiquement avec libstdc\$1\$1 sous Linux. Ils sont censés fonctionner sur toutes les distributions Linux 64 bits qui incluent une glibc version 2.5 ou ultérieure.
Les utilisateurs des anciennes distributions Linux peuvent créer le KPL en utilisant les instructions de compilation fournies avec le code source activé GitHub. Pour télécharger le KPL depuis GitHub, consultez la bibliothèque [Amazon Kinesis](https://github.com/awslabs/amazon-kinesis-producer) Producer.
**Important**
La version 0.x d'Amazon Kinesis Producer Library (KPL) sera disponible le 30 end-of-support janvier 2026. Nous vous **recommandons vivement** de migrer vos applications KPL à l'aide de la version 0.x vers la dernière version de KPL avant le 30 janvier 2026. Pour trouver la dernière version de KPL, consultez la [page KPL sur](https://github.com/awslabs/amazon-kinesis-producer) Github. Pour plus d'informations sur la migration de KPL 0.x vers KPL 1.x, consultez. [Migrer de KPL 0.x vers KPL 1.x](kpl-migration-1x.md)
# Migrer de KPL 0.x vers KPL 1.x
Cette rubrique fournit des step-by-step instructions pour faire migrer votre client de KPL 0.x vers KPL 1.x. KPL 1.x introduit le support de la version AWS SDK pour Java 2.x tout en maintenant la compatibilité de l'interface avec les versions précédentes. Il n'est pas nécessaire de mettre à jour votre logique de traitement des données de base pour migrer vers KPL 1.x.
1. **Assurez-vous de remplir les conditions préalables suivantes :**
+ Kit de développement Java (JDK) 8 ou version ultérieure
+ AWS SDK pour Java 2. x
+ Maven ou Gradle pour la gestion des dépendances
1. **Ajouter des dépendances**
Si vous utilisez Maven, ajoutez la dépendance suivante à votre fichier pom.xml. Assurez-vous d'avoir mis à jour le GroupId de `com.amazonaws` vers `software.amazon.kinesis` et la version `1.x.x` vers la dernière version de KPL.
```
software.amazon.kinesis
amazon-kinesis-producer
1.x.x
```
Si vous utilisez Gradle, ajoutez ce qui suit à votre `build.gradle` fichier. Assurez-vous de le remplacer `1.x.x` par la dernière version de KPL.
```
implementation 'software.amazon.kinesis:amazon-kinesis-producer:1.x.x'
```
Vous pouvez vérifier la dernière version du KPL sur le référentiel [central Maven](https://central.sonatype.com/search?q=amazon-kinesis-producer).
1. **Mettre à jour les déclarations d'importation pour KPL**
Le KPL 1.x utilise le AWS SDK pour Java 2.x et utilise un nom de package mis à jour qui commence par`software.amazon.kinesis`, par rapport au nom du package dans le KPL précédent qui commence par. `com.amazonaws.services.kinesis`
Remplacez l'import pour `com.amazonaws.services.kinesis` par`software.amazon.kinesis`. Le tableau suivant répertorie les importations que vous devez remplacer.
**Importer des pièces de rechange**
[\[See the AWS documentation website for more details\]](http://docs.aws.amazon.com/fr_fr/streams/latest/dev/kpl-migration-1x.html)
1. **Mettre à jour les instructions d'importation pour les classes de fournisseurs AWS d'identifiants**
Lors de la migration vers KPL 1.x, vous devez mettre à jour les packages et les classes de vos importations dans le code de votre application KPL basés sur la version AWS SDK pour Java 1.x vers les packages correspondants basés sur la version 2.x. AWS SDK pour Java Les importations courantes dans l'application KPL sont les classes de fournisseurs d'informations d'identification. Consultez la section [Modifications du fournisseur d'informations d'identification](https://docs.aws.amazon.com/sdk-for-java/latest/developer-guide/migration-client-credentials.html) dans la documentation du guide de migration AWS SDK pour Java 2.x pour obtenir la liste complète des modifications apportées au fournisseur d'informations d'identification. Voici la modification d'importation courante que vous devrez peut-être apporter à vos applications KPL.
**Importer dans KPL 0.x**
```
import com.amazonaws.auth.DefaultAWSCredentialsProviderChain;
```
**Importer dans KPL 1.x**
```
import software.amazon.awssdk.auth.credentials.DefaultCredentialsProvider;
```
Si vous importez d'autres fournisseurs d'informations d'identification basés sur la AWS SDK pour Java version 1.x, vous devez les mettre à jour vers les fournisseurs équivalents AWS SDK pour Java 2.x. Si vous n'en avez importé aucune classes/packages depuis la version AWS SDK pour Java 1.x, vous pouvez ignorer cette étape.
1. **Mettre à jour la configuration du fournisseur d'informations d'identification dans la configuration KPL**
La configuration du fournisseur d'informations d'identification dans KPL 1.x nécessite les fournisseurs d'informations d'identification AWS SDK pour Java 2.x. Si vous transmettez des fournisseurs d'informations d'identification pour la AWS SDK pour Java version 1.x en remplaçant le fournisseur d'informations d'identification `KinesisProducerConfiguration` par défaut, vous devez le mettre à jour avec les fournisseurs d'informations d'identification AWS SDK pour Java 2.x. Consultez la section [Modifications du fournisseur d'informations d'identification](https://docs.aws.amazon.com/sdk-for-java/latest/developer-guide/migration-client-credentials.html) dans la documentation du guide de migration AWS SDK pour Java 2.x pour obtenir la liste complète des modifications apportées au fournisseur d'informations d'identification. Si vous n'avez pas remplacé le fournisseur d'informations d'identification par défaut dans la configuration KPL, vous pouvez ignorer cette étape.
Par exemple, si vous remplacez le fournisseur d'informations d'identification par défaut pour le KPL par le code suivant :
```
KinesisProducerConfiguration config = new KinesisProducerConfiguration();
// SDK v1 default credentials provider
config.setCredentialsProvider(new DefaultAWSCredentialsProviderChain());
```
Vous devez les mettre à jour avec le code suivant pour utiliser le fournisseur d'informations d'identification AWS SDK pour Java 2.x :
```
KinesisProducerConfiguration config = new KinesisProducerConfiguration();
// New SDK v2 default credentials provider
config.setCredentialsProvider(DefaultCredentialsProvider.create());
```
# Transition vers les certificats Amazon Trust Services (ATS) pour le KPL
Le 9 février 2018, à 9 h PST, Amazon Kinesis Data Streams a installé les certificats ATS. Pour continuer à écrire des enregistrements dans Kinesis Data Streams à l'aide de la bibliothèque Amazon Kinesis Producer (KPL), vous devez mettre à niveau votre installation de la KPL vers la version 0.12.6 [ou](http://search.maven.org/#artifactdetails|com.amazonaws|amazon-kinesis-producer|0.12.6|jar) ultérieure. Ce changement concerne toutes les AWS régions.
Pour plus d'informations sur le passage à ATS, voir [Comment préparer AWS le passage à sa propre autorité de certification](https://aws.amazon.com/blogs/security/how-to-prepare-for-aws-move-to-its-own-certificate-authority/).
Si vous rencontrez des problèmes et que vous avez besoin d'une assistance technique, [créez une demande](https://console.aws.amazon.com/support/v1#/case/create) auprès du centre de support AWS .
# Plateformes prises en charge par KPL
La bibliothèque Amazon Kinesis Producer (KPL) est écrite en C\$1\$1 et s'exécute en tant que processus secondaire du processus utilisateur principal. Des fichiers binaires natifs 64 bits précompilés sont regroupés avec la version Java et sont gérés par le wrapper Java.
Le package Java s'exécute sans avoir besoin d'installer les bibliothèques supplémentaires sur les systèmes d'exploitation suivants :
+ Distributions Linux avec le noyau 2.6.18 (septembre 2006) et ultérieur
+ Apple iOS X 10.9 et versions ultérieures
+ Windows Server 2008 et version ultérieure
**Important**
Windows Server 2008 ou version ultérieure est pris en charge pour toutes les versions de KPL jusqu'à la version 0.14.0.
La plateforme Windows n'est pas prise en charge à partir de la version 0.14.0 de KPL.
Notez que la bibliothèque KPL est en 64 bits uniquement.
## Code source
Si les fichiers binaires fournis dans l'installation KPL ne sont pas suffisants pour votre environnement, la base de la KPL est écrite sous forme d'un module C\$1\$1. Le code source du module C\$1\$1 et de l'interface Java est publié sous licence publique Amazon et est disponible sur la bibliothèque GitHub [Amazon Kinesis Producer](https://github.com/awslabs/amazon-kinesis-producer). Bien que la bibliothèque KPL puisse être utilisée sur n'importe quelle plateforme équipée d'un compilateur C\$1\$1 récent en conformité et d'un JRE, Amazon ne prend pas officiellement en charge celles qui ne figurent pas sur la liste des plateformes prises en charge.
# Concepts clés du KPL
Les sections suivantes contiennent les concepts et la terminologie nécessaires pour comprendre et tirer parti de l'Amazon Kinesis Producer Library (KPL).
**Topics**
+ [
## Enregistrements
](#kinesis-kpl-concepts-records)
+ [
## Traitement par lot
](#kinesis-kpl-concepts-batching)
+ [
## Agrégation
](#kinesis-kpl-concepts-aggretation)
+ [
## Collection
](#kinesis-kpl-concepts-collection)
## Enregistrements
Dans ce guide, nous différencions les *enregistrements utilisateur KPL* et les *enregistrements Kinesis Data Streams*. Lorsque nous utilisons le terme *enregistrement* sans identificateur, nous faisons référence à un *enregistrement utilisateur KPL*. Lorsque nous faisons référence à un enregistrement Kinesis Data Streams, il est explicitement question d'enregistrement *Kinesis Data Streams*.
Un enregistrement utilisateur KPL est un blob de données qui a une signification particulière pour l'utilisateur. Les exemples incluent un blob JSON représentant un événement de l'interface utilisateur sur un site Web, ou une entrée de journal issue d'un serveur Web.
Un enregistrement Kinesis Data Streams est une instance de la structure de données `Record` définie par l'API du service Kinesis Data Streams. Il contient une clé de partition, un numéro de séquence et un blob de données.
## Traitement par lot
Le *traitement par lot* consiste à exécuter une action sur plusieurs éléments au lieu d'exécuter cette action à plusieurs reprises sur chaque élément individuel.
Dans ce contexte, l'« élément » est un enregistrement et l'action est l'envoi de cet enregistrement à Kinesis Data Streams. Dans une situation sans traitement par lot, vous placez chaque enregistrement dans un enregistrement Kinesis Data Streams distinct et effectuez une seule demande HTTP pour envoyer l'enregistrement à Kinesis Data Streams. Avec le traitement par lot, chaque demande HTTP peut exécuter plusieurs enregistrements au lieu d'un seul.
La KPL prend en charge deux types de traitement par lot :
+ *Regroupement* : stockage de plusieurs enregistrements dans un seul enregistrement Kinesis Data Streams.
+ *Collecte* : utilisation de l'opération d'API `PutRecords` pour envoyer plusieurs enregistrements Kinesis Data Streams dans une ou plusieurs partitions de votre flux de données Kinesis.
Les deux types de traitement par lot KPL sont conçus pour co-exister et peuvent être activés ou désactivés indépendamment l'un de l'autre. Par défaut, les deux types sont activés.
## Agrégation
Le *regroupement* consiste à stocker plusieurs enregistrements dans un enregistrement Kinesis Data Streams. Le regroupement permet aux clients d'augmenter le nombre d'enregistrements envoyés par appel d'API, ce qui augmente efficacement le lent e documents envoyés par l'appel d'API, ce qui augmente efficacement le débit de l'application producteur.
Les partitions Kinesis Data Streams acceptent jusqu'à 1 000 enregistrements Kinesis Data Streams par seconde, soit un débit de 1 Mo. Le nombre limite d'enregistrements Kinesis Data Streams par seconde force les clients à avoir des enregistrements inférieurs à 1 Ko. Le regroupement des enregistrements permet aux clients de combiner plusieurs enregistrements en un seul enregistrement Kinesis Data Streams. Cela permet aux clients d'améliorer leur débit par partition.
Prenons le cas d'une partition de la région us-east-1 qui fonctionne actuellement à un rythme constant de 1 000 enregistrements par seconde, avec des enregistrements de 512 octets chacun. Avec le regroupement KPL, vous pouvez compresser 1 000 enregistrements en seulement 10 enregistrements Kinesis Data Streams, ce qui réduit le RPS à 10 (de 50 Ko chacun).
## Collection
La *collecte* consiste à traiter par lot plusieurs enregistrements Kinesis Data Streams et à les envoyer dans une seule demande HTTP avec un appel d'opération API `PutRecords`, au lieu d'envoyer chaque enregistrement Kinesis Data Streams dans sa propre demande HTTP.
Cela augmente le débit en réduisant le coût lié aux nombreuses demandes HTTP distinctes. En fait, l'API `PutRecords` elle-même a été spécifiquement conçue à cet effet.
La collecte est différente du regroupement en cela qu'elle utilise des groupes d'enregistrements Kinesis Data Streams. Les enregistrements Kinesis Data Streams collectés peuvent encore contenir plusieurs enregistrements provenant de l'utilisateur. La relation peut être visualisée comme suit :
```
record 0 --|
record 1 | [ Aggregation ]
... |--> Amazon Kinesis record 0 --|
... | |
record A --| |
|
... ... |
|
record K --| |
record L | | [ Collection ]
... |--> Amazon Kinesis record C --|--> PutRecords Request
... | |
record S --| |
|
... ... |
|
record AA--| |
record BB | |
... |--> Amazon Kinesis record M --|
... |
record ZZ--|
```
# Intégrer le KPL au code du producteur
La bibliothèque Amazon Kinesis Producer (KPL) s'exécute dans le cadre d'un processus distinct et communique avec votre processus utilisateur parent à l'aide d'IPC. Cette architecture est parfois appelée un [microservice](http://en.wikipedia.org/wiki/Microservices) et est choisie pour deux raisons principales :
**1) Votre processus utilisateur ne se bloque pas même si la KPL se bloque**
Votre processus peut avoir des tâches non liées à Kinesis Data Streams et peuvent être en mesure de poursuivre l'opération même si la KPL se bloque. Votre processus utilisateur parent peut redémarrer la KPL et redevenir entièrement opérationnel (cette fonctionnalité existe dans les encapsuleurs officiels).
Prenons l'exemple d'un serveur Web qui envoie des métriques à Kinesis Data Streams. Le serveur peut continuer à servir des pages même si la partie Kinesis Data Streams a cessé de fonctionner. Le blocage du serveur entier suite à un bogue dans la KPL provoque donc une panne inutile.
**2) Les clients arbitraires peuvent être pris en charge**
Il y a toujours des clients qui utilisent des langages autres que ceux qui sont officiellement pris en charge. Ces clients doivent également pouvoir utiliser la KPL facilement.
## Matrice d'utilisation recommandée
La matrice d'utilisation suivante répertorie les paramètres recommandés pour les différents utilisateurs et vous indique si et comment vous devez utiliser le KPL. N'oubliez pas que si le regroupement est activé, le dégroupement doit aussi être utilisé pour extraire vos enregistrements du côté consommateur.
| Langage côté producteur | Langage côté consommateur | Version de la KCL | Logique de contrôle | Pouvez-vous utiliser la KPL ? | Mises en garde |
| --- | --- | --- | --- | --- | --- |
| Tout sauf Java | \$1 | \$1 | \$1 | Non | N/A |
| Java | Java | Utilise le SDK Java directement | N/A | Oui | Si le regroupement est utilisé, vous devez utiliser la bibliothèque de regroupement fournie après les appels de GetRecords. |
| Java | Tout sauf Java | Utilise le SDK directement | N/A | Oui | Désactiver obligatoirement le regroupement. |
| Java | Java | 1.3.x | N/A | Oui | Désactiver obligatoirement le regroupement. |
| Java | Java | 1.4.x | Appelle le contrôle sans argument | Oui | Aucune |
| Java | Java | 1.4.x | Appelle le contrôle avec un numéro de séquence explicite | Oui | Désactiver le regroupement ou modifier le code pour utiliser des numéros de séquence étendus pour le contrôle. |
| Java | Tout sauf Java | 1.3.x \$1 démon multilingue \$1 wrapper propre au langage | N/A | Oui | Désactiver obligatoirement le regroupement. |
# Écrivez dans votre flux de données Kinesis à l'aide du KPL
Les sections suivantes présentent des exemples de code dans une progression allant du code de production le plus élémentaire au code entièrement asynchrone.
## Code du producteur Barebones
Le code suivant contient tous les éléments nécessaires pour écrire une application producteur convenable minimale. Les enregistrements utilisateur de l'Amazon Kinesis Producer Library (KPL) sont traités en arrière-plan.
```
// KinesisProducer gets credentials automatically like
// DefaultAWSCredentialsProviderChain.
// It also gets region automatically from the EC2 metadata service.
KinesisProducer kinesis = new KinesisProducer();
// Put some records
for (int i = 0; i < 100; ++i) {
ByteBuffer data = ByteBuffer.wrap("myData".getBytes("UTF-8"));
// doesn't block
kinesis.addUserRecord("myStream", "myPartitionKey", data);
}
// Do other stuff ...
```
## Réagir aux résultats de manière synchrone
Dans l'exemple précédent, le code n'a pas vérifié si les enregistrements KPL utilisateur ont abouti. La KPL effectue toutes les tentatives nécessaires pour tenir compte des défaillances. Si vous voulez vérifier les résultats, vous pouvez les examiner à l'aide des objets `Future` renvoyés par `addUserRecord`, comme dans l'exemple suivant (exemple précédent indiqué pour le contexte) :
```
KinesisProducer kinesis = new KinesisProducer();
// Put some records and save the Futures
List> putFutures = new LinkedList>();
for (int i = 0; i < 100; i++) {
ByteBuffer data = ByteBuffer.wrap("myData".getBytes("UTF-8"));
// doesn't block
putFutures.add(
kinesis.addUserRecord("myStream", "myPartitionKey", data));
}
// Wait for puts to finish and check the results
for (Future f : putFutures) {
UserRecordResult result = f.get(); // this does block
if (result.isSuccessful()) {
System.out.println("Put record into shard " +
result.getShardId());
} else {
for (Attempt attempt : result.getAttempts()) {
// Analyze and respond to the failure
}
}
}
```
## Répondre aux résultats de manière asynchrone
L'exemple précédent fait appel `get()` à un `Future` objet qui bloque l'exécution. Si vous ne souhaitez pas bloquer l'exécution, vous pouvez utiliser un rappel asynchrone, comme illustré dans l'exemple suivant :
```
KinesisProducer kinesis = new KinesisProducer();
FutureCallback myCallback = new FutureCallback() {
@Override public void onFailure(Throwable t) {
/* Analyze and respond to the failure */
};
@Override public void onSuccess(UserRecordResult result) {
/* Respond to the success */
};
};
for (int i = 0; i < 100; ++i) {
ByteBuffer data = ByteBuffer.wrap("myData".getBytes("UTF-8"));
ListenableFuture f = kinesis.addUserRecord("myStream", "myPartitionKey", data);
// If the Future is complete by the time we call addCallback, the callback will be invoked immediately.
Futures.addCallback(f, myCallback);
}
```
# Configuration de la bibliothèque Amazon Kinesis Producer
Bien que les paramètres par défaut fonctionnent en principe bien pour la plupart des cas d'utilisation, vous pouvez en modifier certains paramètres par défaut pour adapter le comportement de `KinesisProducer` à vos besoins. Pour cela, une instance de la classe `KinesisProducerConfiguration` peut être passée au constructeur `KinesisProducer`, par exemple :
```
KinesisProducerConfiguration config = new KinesisProducerConfiguration()
.setRecordMaxBufferedTime(3000)
.setMaxConnections(1)
.setRequestTimeout(60000)
.setRegion("us-west-1");
final KinesisProducer kinesisProducer = new KinesisProducer(config);
```
Vous pouvez également charger une configuration à partir d'un fichier de propriétés :
```
KinesisProducerConfiguration config = KinesisProducerConfiguration.fromPropertiesFile("default_config.properties");
```
Vous pouvez remplacer un chemin d'accès et un nom de fichier auxquels le processus de l'utilisateur peut accéder. Vous pouvez aussi appeler les méthodes set sur l'instance `KinesisProducerConfiguration` créée de cette façon afin de personnaliser la configuration.
Le fichier de propriétés doit spécifier les paramètres en utilisant leur nom dans PascalCase. Les noms correspondent à ceux qui sont utilisés dans les méthodes set de la classe `KinesisProducerConfiguration`. Par exemple :
```
RecordMaxBufferedTime = 100
MaxConnections = 4
RequestTimeout = 6000
Region = us-west-1
```
Pour plus d'informations sur les règles d'utilisation des paramètres de configuration et les limites de valeurs, consultez l'[exemple de fichier de propriétés de configuration sur GitHub](https://github.com/awslabs/amazon-kinesis-producer/blob/master/java/amazon-kinesis-producer-sample/default_config.properties).
Notez que, une fois que `KinesisProducer` est initialisé, la modification de l'instance `KinesisProducerConfiguration` qui a été utilisée n'a aucun autre effet. `KinesisProducer` ne soutient pas actuellement la reconfiguration dynamique.
# Mettre en œuvre la désagrégation des consommateurs
À partir de la version 1.4.0, la KCL prend en charge le dégroupement automatique des enregistrements utilisateur KPL. Le code de l'application consommateur écrit avec les versions antérieures de la KCL sera compilé sans aucune modification lorsque vous aurez mis à jour la KCL. Toutefois, si la fonctionnalité de regroupement KPL est utilisée côté producteur, le contrôle comporte une subtilité : tous les sous-enregistrements figurant dans un enregistrement ont le même numéro de séquence si bien que des données supplémentaires doivent être stockées avec le point de contrôle si vous avez besoin de différencier les sous-enregistrements. Ces données supplémentaires s'appellent le *numéro de sous-séquence*.
**Topics**
+ [
## Migrer depuis les versions précédentes de la KCL
](#kinesis-kpl-consumer-deaggregation-migration)
+ [
## Utiliser les extensions KCL pour la désagrégation KPL
](#kinesis-kpl-consumer-deaggregation-extensions)
+ [
## Utiliser GetRecords directement
](#kinesis-kpl-consumer-deaggregation-getrecords)
## Migrer depuis les versions précédentes de la KCL
Vous n'êtes pas obligé de modifier vos appels existants pour effectuer des points de contrôle avec agrégation. L'extraction de tous les enregistrements stockés avec succès dans Kinesis Data Streams demeure garantie. La KCL propose désormais deux nouvelles opérations de point de contrôle pour prendre en charge des cas d'utilisation particuliers, décrits ci-dessous.
Si votre code existant a été écrit pour la KCL avant le support KPL et que votre opération de point de contrôle est appelée sans arguments, cela revient à vérifier le numéro de séquence du dernier enregistrement utilisateur KPL du lot. Si votre opération de contrôle est appelée avec une chaîne de numéro de séquence, cela équivaut à contrôler le numéro de séquence donné du lot ainsi que le numéro de sous-séquence implicite 0 (zéro).
L'appel de la nouvelle opération de contrôle KCL `checkpoint()` sans argument est équivalent sémantiquement au contrôle du numéro de séquence du dernier appel `Record` du lot, accompagné du numéro de sous-séquence implicite 0 (zéro).
L'appel de la nouvelle opération de contrôle KCL `checkpoint(Record record)` est équivalent sémantiquement au contrôle du numéro de séquence donné du `Record`, accompagné du numéro de sous-séquence implicite 0 (zéro). Si l'appel `Record` est en fait un `UserRecord`, le numéro de séquence et le numéro de sous-séquence `UserRecord` sont contrôlés.
L'appel de la nouvelle opération de contrôle KCL `checkpoint(String sequenceNumber, long subSequenceNumber)` contrôle explicitement le numéro de séquence donné ainsi que le numéro de sous-séquence donné.
Dans chacun de ces cas, une fois que le point de contrôle est stocké dans la table de point de contrôle Amazon DynamoDB, la KCL peut correctement reprendre l'extraction des enregistrements même lorsque l'application se bloque et redémarre. Si la séquence contient plusieurs enregistrements, l'extraction commence par l'enregistrement ayant le numéro de sous-séquence suivant dans l'enregistrement ayant le numéro de séquence contrôlé le plus récemment. Si le dernier contrôle incluait le tout dernier numéro de sous-séquence de l'enregistrement ayant le numéro de séquence précédent, l'extraction commence par l'enregistrement avec le numéro de séquence suivant.
La section suivante décrit en détail les points de contrôle des séquences et des sous-séquences pour les consommateurs, qui doivent éviter de sauter ou de dupliquer des enregistrements. S'il importe peu que des enregistrements soient sautés (ou dupliqués) lors de l'arrêt ou du redémarrage du traitement des enregistrements du consommateur, vous pouvez exécuter votre code existant sans modification.
## Utiliser les extensions KCL pour la désagrégation KPL
La désagrégation KPL peut impliquer un point de contrôle des sous-séquences. Pour faciliter l'utilisation du contrôle de sous-séquence, une classe `UserRecord` a été ajoutée à la KCL :
```
public class UserRecord extends Record {
public long getSubSequenceNumber() {
/* ... */
}
@Override
public int hashCode() {
/* contract-satisfying implementation */
}
@Override
public boolean equals(Object obj) {
/* contract-satisfying implementation */
}
}
```
Cette classe est maintenant utilisée à la place de `Record`. Elle ne casse pas le code existant, car c'est une sous-classe de `Record`. La classe `UserRecord` représente à la fois les sous-enregistrements réels et les enregistrements non regroupés standard. Les enregistrements non regroupés peuvent être considérés comme des enregistrements regroupés contenant un seul sous-enregistrement.
En outre, deux nouvelles opérations ont été ajoutées à `IRecordProcessorCheckpointer` :
```
public void checkpoint(Record record);
public void checkpoint(String sequenceNumber, long subSequenceNumber);
```
Pour commencer à utiliser le contrôle du numéro de sous-séquence, vous pouvez effectuer la conversion ci-dessous. Modifiez le code de formulaire suivant :
```
checkpointer.checkpoint(record.getSequenceNumber());
```
Nouveau code de formulaire :
```
checkpointer.checkpoint(record);
```
Nous vous recommandons d'utiliser le formulaire `checkpoint(Record record)` pour le contrôle de sous-séquence. Toutefois, si vous stockez déjà `sequenceNumbers` dans les chaînes à utiliser pour les points de contrôle, vous devez désormais stocker aussi `subSequenceNumber`, comme l'illustre l'exemple suivant :
```
String sequenceNumber = record.getSequenceNumber();
long subSequenceNumber = ((UserRecord) record).getSubSequenceNumber(); // ... do other processing
checkpointer.checkpoint(sequenceNumber, subSequenceNumber);
```
Le cast from `Record` to réussit `UserRecord` toujours car l'implémentation utilise `UserRecord` toujours. À moins qu'il ne soit nécessaire d'effectuer des opérations arithmétiques sur les numéros de séquence, cette approche n'est pas recommandée.
Lors du traitement des enregistrements utilisateur KPL, la KCL écrit le numéro de sous-séquence dans Amazon DynamoDB sous la forme d'un champ supplémentaire pour chaque ligne. Les versions antérieures de la KCL utilisaient `AFTER_SEQUENCE_NUMBER` pour extraire les enregistrements lors de la reprise des points de contrôle. La KCL actuelle avec prise en charge KPL utilise `AT_SEQUENCE_NUMBER` à la place. Lorsque l'enregistrement situé au numéro de séquence contrôlé est extrait, le numéro de sous-séquence contrôlé est vérifié, et les sous-enregistrements sont abandonnés le cas échéant (ce peut être tous si le dernier sous-enregistrement est celui qui est contrôlé). Encore, les enregistrements non regroupés peuvent être considérés comme des enregistrements regroupés contenant un seul sous-enregistrement, si bien que le même algorithme fonctionne aussi bien pour les enregistrements regroupés que pour les enregistrements non regroupés.
## Utiliser GetRecords directement
Vous pouvez également choisir de ne pas utiliser la KCL, mais d'invoquer directement l'opération d'API `GetRecords` pour extraire les enregistrements Kinesis Data Streams. Pour décompresser ces enregistrements dans vos enregistrements utilisateur KPL initiaux, appelez l'une des opérations statiques suivantes figurant dans `UserRecord.java` :
```
public static List deaggregate(List records)
public static List deaggregate(List records, BigInteger startingHashKey, BigInteger endingHashKey)
```
La première opération utilise la valeur par défaut `0` (zéro) pour `startingHashKey` et la valeur par défaut `2^128 -1` pour `endingHashKey`.
Chacune de ces opérations dégroupe la liste des enregistrements Kinesis Data Streams dans une liste d'enregistrements utilisateur KPL. Tout enregistrement utilisateur KPL dont la clé de hachage ou la clé de partition explicite n'est pas comprise dans la plage de `startingHashKey` (inclus) et de `endingHashKey` (inclus) est supprimé de la liste d'enregistrements renvoyée.
# Utiliser le KPL avec Amazon Data Firehose
Si vous utilisez la bibliothèque producteur Kinesis (KPL) pour écrire des données dans un flux de données Kinesis, vous pouvez utiliser un regroupement pour combiner ces enregistrements Kinesis. Si vous utilisez ensuite ce flux de données comme source pour votre flux de diffusion Firehose, Firehose désagrège les enregistrements avant de les livrer à destination. Si vous configurez votre flux de diffusion pour transformer les données, Firehose désagrège les enregistrements avant de les transmettre. AWS Lambda Pour plus d'informations, consultez [Écriture de données à Amazon Kinesis Firehose à l'aide de flux de données Kinesis](https://docs.aws.amazon.com/firehose/latest/dev/writing-with-kinesis-streams.html) (français non garanti).
# Utiliser le KPL avec le registre des AWS Glue schémas
Vous pouvez intégrer vos flux de données Kinesis au registre des AWS Glue schémas. Le registre des AWS Glue schémas vous permet de découvrir, de contrôler et de faire évoluer les schémas de manière centralisée, tout en garantissant que les données produites sont continuellement validées par un schéma enregistré. Un schéma définit la structure et le format d'un enregistrement de données. Un schéma est une spécification versionnée pour la publication, la consommation ou le stockage des données fiables. Le registre des AWS Glue schémas vous permet d'améliorer la qualité end-to-end des données et la gouvernance des données au sein de vos applications de streaming. Pour plus d'informations, consultez le registre [AWS Glue Schema](https://docs.aws.amazon.com/glue/latest/dg/schema-registry.html) (français non garanti). Vous pouvez notamment configurer cette intégration via les bibliothèques KPL et KCL en Java.
**Important**
Actuellement, l'intégration de Kinesis Data Streams AWS Glue et de registres de schémas n'est prise en charge que pour les flux de données Kinesis qui utilisent des producteurs KPL implémentés en Java. La prise en charge multilingue n'est pas fournie.
Pour obtenir des instructions détaillées sur la façon de configurer l'intégration de Kinesis Data Streams à Schema Registry à l'aide du KPL, consultez la section « Interaction avec les données à KPL/KCL l'aide des bibliothèques » [dans Use Case : Integrating Amazon Kinesis Data Streams with AWS the](https://docs.aws.amazon.com/glue/latest/dg/schema-registry-integrations.html#schema-registry-integrations-kds) Glue Schema Registry.
# Configurer la configuration du proxy KPL
Pour les applications qui ne peuvent pas se connecter directement à Internet, tous les clients du AWS SDK prennent en charge l'utilisation de proxys HTTP ou HTTPS. Au sein d'un environnement d'entreprise classique, tout le trafic réseau sortant doit passer par des serveurs proxy. Si votre application utilise la Kinesis Producer Library (KPL) pour collecter et envoyer des données AWS dans un environnement qui utilise des serveurs proxy, elle doit être configurée en tant que proxy KPL. KPL est une bibliothèque de haut niveau basée sur le SDK AWS Kinesis. Elle est divisée en un processus natif et un encapsuleur. Le processus natif exécute toutes les tâches de traitement et d'envoi des enregistrements, tandis que l'encapsuleur gère le processus natif et communique avec lui. Pour obtenir plus d'informations, consultez [Implémentation de producteurs efficaces et fiables avec la bibliothèque producteur Amazon Kinesis](https://aws.amazon.com/blogs/big-data/implementing-efficient-and-reliable-producers-with-the-amazon-kinesis-producer-library/) (français non garanti).
L'encapsuleur est écrit en Java et le processus natif est écrit en C\$1\$1 à l'aide du SDK Kinesis. Les versions 0.14.7 et supérieures de la KPL prennent désormais en charge la configuration de proxy dans l'encapsuleur Java qui peut transmettre toutes les configurations de proxy au processus natif. Pour plus d'informations, consultez [https://github.com/awslabs/amazon-kinesis-producer/releases/tag/v0.14.7](https://github.com/awslabs/amazon-kinesis-producer/releases/tag/v0.14.7).
Utilisez le code suivant pour ajouter des configurations de proxy à vos applications KPL.
```
KinesisProducerConfiguration configuration = new KinesisProducerConfiguration();
// Next 4 lines used to configure proxy
configuration.setProxyHost("10.0.0.0"); // required
configuration.setProxyPort(3128); // default port is set to 443
configuration.setProxyUserName("username"); // no default
configuration.setProxyPassword("password"); // no default
KinesisProducer kinesisProducer = new KinesisProducer(configuration);
```
# Politique de cycle de vie des versions KPL
Cette rubrique décrit la politique de cycle de vie des versions pour Amazon Kinesis Producer Library (KPL). AWS fournit régulièrement de nouvelles versions pour les versions de KPL afin de prendre en charge les nouvelles fonctionnalités et améliorations, les corrections de bogues, les correctifs de sécurité et les mises à jour de dépendances. Nous vous recommandons de vous en tenir up-to-date aux versions de KPL pour vous tenir au courant des dernières fonctionnalités, des mises à jour de sécurité et des dépendances sous-jacentes. Nous **ne recommandons pas** de continuer à utiliser une version de KPL non prise en charge.
Le cycle de vie des principales versions de KPL comprend les trois phases suivantes :
+ **Disponibilité générale (GA)** — Au cours de cette phase, la version majeure est entièrement prise en charge. AWS fournit régulièrement des versions mineures et des correctifs qui incluent la prise en charge de nouvelles fonctionnalités ou des mises à jour d'API pour Kinesis Data Streams, ainsi que des correctifs de bogues et de sécurité.
+ **Mode maintenance** : AWS limite les versions de correctifs publiées uniquement pour résoudre les corrections de bogues critiques et les problèmes de sécurité. La version majeure ne recevra pas de mises à jour relatives aux nouvelles fonctionnalités ou APIs à Kinesis Data Streams.
+ **E nd-of-support** — La version majeure ne recevra plus de mises à jour ni de versions. Les versions précédemment publiées continueront d'être disponibles via les gestionnaires de packages publics et le code restera activé GitHub. L'utilisation d'une version atteinte se end-of-support fait à la discrétion de l'utilisateur. Nous vous recommandons de passer à la dernière version majeure.
| Version majeure | Phase en cours | Date de publication | Date du mode de maintenance | End-of-support date |
| --- | --- | --- | --- | --- |
| KPL 0.x | Mode de maintenance | 02/06/2015 | 17/04/2025 | 30-01-2026/ |
| KPL 1.x | Disponibilité générale | 15/12/2024 | -- | -- |
# Développez des producteurs à l'aide de l'API Amazon Kinesis Data Streams avec le AWS SDK pour Java
Vous pouvez développer des producteurs à l'aide de l'API Amazon Kinesis Data Streams associée AWS au SDK pour Java. Si vous ne connaissez pas Kinesis Data Streams, commencez par vous familiariser avec les concepts et la terminologie présentés dans [Qu'est-ce qu'Amazon Kinesis Data Streams ?](introduction.md) et [Utilisez le AWS CLI pour effectuer des opérations Amazon Kinesis Data Streams](getting-started.md).
Ces exemples traitent de l'[API Kinesis Data Streams](https://docs.aws.amazon.com/kinesis/latest/APIReference/) et utilisent le [SDK AWS pour Java](https://aws.amazon.com/sdk-for-java/) afin d'ajouter (mettre) des données dans un flux. Cependant, pour la plupart des cas d'utilisation, vous préférerez sans doute la bibliothèque Kinesis Data Streams. Pour de plus amples informations, veuillez consulter [Développez des producteurs à l'aide de la bibliothèque Amazon Kinesis Producer Library (KPL)](developing-producers-with-kpl.md).
L'exemple de code Java présenté dans ce chapitre montre comment effectuer les opérations de base d'API Kinesis Data Streams et est divisé logiquement par type d'opération. Ces exemples ne représentent pas du code prêt à la production, car ils ne recherchent pas toutes les exceptions possibles ou ne tiennent pas compte de toutes les considérations possibles en matière de sécurité ou de performances. En outre, vous pouvez appeler [l'API Kinesis Data Streams](https://docs.aws.amazon.com/kinesis/latest/APIReference/) à l'aide d'autres langages de programmation. Pour plus d'informations sur toutes les options disponibles AWS SDKs, consultez [Commencer à développer avec Amazon Web Services](https://aws.amazon.com/developers/getting-started/).
Chaque tâche a des conditions préalables ; par exemple, vous ne pouvez pas ajouter de données à un flux tant que vous n'avez pas créé de flux, ce qui demande de créer un client. Pour de plus amples informations, veuillez consulter [Création et gestion de flux de données Kinesis](working-with-streams.md).
**Topics**
+ [
## Ajouter des données à un flux
](#kinesis-using-sdk-java-add-data-to-stream)
+ [
# Interagissez avec les données à l'aide du registre des AWS Glue schémas
](kinesis-integration-glue-schema-registry.md)
## Ajouter des données à un flux
Une fois qu'un flux est créé, vous pouvez y ajouter des données sous forme d'enregistrements. Un enregistrement est une structure de données qui contient les données à traiter sous la forme d'un blob de données. Une fois que vous avez stocké les données dans l'enregistrement, Kinesis Data Streams n'inspecte pas, n'interprète pas ou ne modifie absolument pas les données. Chaque enregistrement a également un numéro de séquence et une clé de partition qui lui sont associés.
L'API Kinesis Data Streams comporte deux opérations différentes qui ajoutent des données à un flux, [https://docs.aws.amazon.com/kinesis/latest/APIReference/API_PutRecords.html](https://docs.aws.amazon.com/kinesis/latest/APIReference/API_PutRecords.html) et [https://docs.aws.amazon.com/kinesis/latest/APIReference/API_PutRecord.html](https://docs.aws.amazon.com/kinesis/latest/APIReference/API_PutRecord.html). L'opération `PutRecords` envoie plusieurs enregistrements à votre flux par demande HTTP et l'opération `PutRecord` envoie des enregistrements à votre flux un à la fois (une demande HTTP distincte est nécessaire pour chaque enregistrement). Vous préférerez sans doute utiliser `PutRecords` pour la plupart des applications, car cette opération permet d'atteindre un débit supérieur par application producteur. Pour plus d'informations sur chacune de ces opérations, consultez les sous-sections distinctes ci-dessous.
**Topics**
+ [
### Ajoutez plusieurs enregistrements avec PutRecords
](#kinesis-using-sdk-java-putrecords)
+ [
### Ajoutez un seul enregistrement avec PutRecord
](#kinesis-using-sdk-java-putrecord)
Étant donné que votre application source ajoute des données au flux à l'aide de l'API Kinesis Data Streams, n'oubliez jamais qu'une ou plusieurs applications consommateur traitent très probablement simultanément des données provenant du flux. Pour plus d'informations sur la façon dont les applications consommateur obtiennent les données à l'aide de l'API Kinesis Data Streams, consultez la page [Obtenir des données à partir d'un flux](developing-consumers-with-sdk.md#kinesis-using-sdk-java-get-data).
**Important**
[Modifier la période de conservation des données](kinesis-extended-retention.md)
### Ajoutez plusieurs enregistrements avec PutRecords
L'opération [https://docs.aws.amazon.com/kinesis/latest/APIReference/API_PutRecords.html](https://docs.aws.amazon.com/kinesis/latest/APIReference/API_PutRecords.html) envoie plusieurs enregistrements à Kinesis Data Streams dans une seule demande. En utilisant `PutRecords`, les applications producteur peuvent atteindre un débit supérieur lors de l'envoi de données à leur flux de données Kinesis. Chaque demande `PutRecords` peut prendre en charge jusqu'à 500 enregistrements. Chaque enregistrement de la demande peut atteindre 1 Mo, jusqu'à une limite de 5 Mo pour l'ensemble de la demande, y compris les clés de partition. Comme avec la seule opération `PutRecord` décrite ci-dessous, `PutRecords` utilise des numéros de séquence et des clés de partition. Toutefois, le paramètre `PutRecord` `SequenceNumberForOrdering` n'est pas inclus dans un appel `PutRecords`. L'opération `PutRecords` tente de traiter tous les enregistrements dans l'ordre naturel de la demande.
Chaque enregistrement de données a un numéro de séquence unique. Le numéro de séquence est attribué par Kinesis Data Streams une fois que vous avez appelé `client.putRecords` pour ajouter les enregistrements de données au flux. Les numéros de séquence correspondant à une même clé de partition deviennent généralement de plus en plus longs au fil du temps ; plus l'intervalle de temps entre chaque demande `PutRecords` est élevé, plus les numéros de séquence sont longs.
**Note**
Les numéros de séquence ne peuvent pas servir d'index aux ensembles de données d'un même flux. Pour séparer logiquement les ensembles de données, utilisez des clés de partition ou créez un flux distinct pour chaque ensemble de données.
Une demande `PutRecords` peut inclure des enregistrements ayant différentes clés de partition. La portée de la demande est un flux ; chaque demande peut inclure une combinaison de clés de partition et d'enregistrements allant jusqu'aux limites définies pour la demande. Les demandes effectuées avec de nombreuses clés de partition différentes pour des flux comportant de nombreuses partitions différentes sont généralement plus rapides que les demandes comportant un petit nombre de clés de partition pour un petit nombre de partitions. Le nombre de clés de partition doit être beaucoup plus grand que le nombre de partitions pour réduire la latence et optimiser le débit.
#### PutRecords exemple
Le code suivant crée 100 enregistrements de données avec des clés de partition séquentielles et les place dans un flux appelé `DataStream`.
```
AmazonKinesisClientBuilder clientBuilder = AmazonKinesisClientBuilder.standard();
clientBuilder.setRegion(regionName);
clientBuilder.setCredentials(credentialsProvider);
clientBuilder.setClientConfiguration(config);
AmazonKinesis kinesisClient = clientBuilder.build();
PutRecordsRequest putRecordsRequest = new PutRecordsRequest();
putRecordsRequest.setStreamName(streamName);
List putRecordsRequestEntryList = new ArrayList<>();
for (int i = 0; i < 100; i++) {
PutRecordsRequestEntry putRecordsRequestEntry = new PutRecordsRequestEntry();
putRecordsRequestEntry.setData(ByteBuffer.wrap(String.valueOf(i).getBytes()));
putRecordsRequestEntry.setPartitionKey(String.format("partitionKey-%d", i));
putRecordsRequestEntryList.add(putRecordsRequestEntry);
}
putRecordsRequest.setRecords(putRecordsRequestEntryList);
PutRecordsResult putRecordsResult = kinesisClient.putRecords(putRecordsRequest);
System.out.println("Put Result" + putRecordsResult);
```
La réponse `PutRecords` comprend un tableau de réponse `Records`. Chaque enregistrement compris dans ce tableau de réponse correspond directement à un enregistrement dans le tableau de demande. Ces entrées sont classées dans l'ordre naturel, soit de haut en bas de la demande et de la réponse. Le tableau de réponse `Records` comprend toujours le même nombre d'enregistrements que le tableau de demande.
#### Gérez les défaillances lors de l'utilisation PutRecords
Par défaut, la défaillance d'enregistrements individuels dans une demande n'arrête pas le traitement des enregistrements suivants dans une demande `PutRecords`. Cela signifie qu'un tableau `Records` de réponse comprend à la fois des enregistrements traités avec succès et sans succès. Vous devez détecter les enregistrements traités sans succès et les inclure dans un appel ultérieur.
Les enregistrements qui ont réussi incluent les valeurs `SequenceNumber` et `ShardID`. Ceux qui ont échoué incluent les valeurs `ErrorCode` et `ErrorMessage`. Le paramètre `ErrorCode` reflète le type d'erreur et peut avoir une des valeurs suivantes : `ProvisionedThroughputExceededException` ou `InternalFailure`. `ErrorMessage` fournit des informations plus détaillées sur l'exception `ProvisionedThroughputExceededException`, y compris l'ID de compte, le nom du flux et les ID de partition de l'enregistrement qui a été limité. L'exemple ci-dessous contient trois enregistrements dans une demande `PutRecords`. Le second enregistrement échoue et est reflété dans la réponse.
**Example PutRecords Syntaxe de la demande**
```
{
"Records": [
{
"Data": "XzxkYXRhPl8w",
"PartitionKey": "partitionKey1"
},
{
"Data": "AbceddeRFfg12asd",
"PartitionKey": "partitionKey1"
},
{
"Data": "KFpcd98*7nd1",
"PartitionKey": "partitionKey3"
}
],
"StreamName": "myStream"
}
```
**Example PutRecords Syntaxe de réponse**
```
{
"FailedRecordCount”: 1,
"Records": [
{
"SequenceNumber": "21269319989900637946712965403778482371",
"ShardId": "shardId-000000000001"
},
{
“ErrorCode":”ProvisionedThroughputExceededException”,
“ErrorMessage": "Rate exceeded for shard shardId-000000000001 in stream exampleStreamName under account 111111111111."
},
{
"SequenceNumber": "21269319989999637946712965403778482985",
"ShardId": "shardId-000000000002"
}
]
}
```
Les enregistrements qui ont été traités sans succès peuvent être inclus dans des demandes `PutRecords` ultérieures. Tout d'abord, vérifiez le paramètre `FailedRecordCount` de `putRecordsResult` afin de savoir si la demande comporte des enregistrements d'échecs. Dans ce cas, chaque `putRecordsEntry` comportant un `ErrorCode` qui n'est pas `null` doit être ajouté à une demande ultérieure. Pour un exemple de ce type de gestionnaire, reportez-vous au code suivant.
**Example PutRecords gestionnaire de défaillances**
```
PutRecordsRequest putRecordsRequest = new PutRecordsRequest();
putRecordsRequest.setStreamName(myStreamName);
List putRecordsRequestEntryList = new ArrayList<>();
for (int j = 0; j < 100; j++) {
PutRecordsRequestEntry putRecordsRequestEntry = new PutRecordsRequestEntry();
putRecordsRequestEntry.setData(ByteBuffer.wrap(String.valueOf(j).getBytes()));
putRecordsRequestEntry.setPartitionKey(String.format("partitionKey-%d", j));
putRecordsRequestEntryList.add(putRecordsRequestEntry);
}
putRecordsRequest.setRecords(putRecordsRequestEntryList);
PutRecordsResult putRecordsResult = amazonKinesisClient.putRecords(putRecordsRequest);
while (putRecordsResult.getFailedRecordCount() > 0) {
final List failedRecordsList = new ArrayList<>();
final List putRecordsResultEntryList = putRecordsResult.getRecords();
for (int i = 0; i < putRecordsResultEntryList.size(); i++) {
final PutRecordsRequestEntry putRecordRequestEntry = putRecordsRequestEntryList.get(i);
final PutRecordsResultEntry putRecordsResultEntry = putRecordsResultEntryList.get(i);
if (putRecordsResultEntry.getErrorCode() != null) {
failedRecordsList.add(putRecordRequestEntry);
}
}
putRecordsRequestEntryList = failedRecordsList;
putRecordsRequest.setRecords(putRecordsRequestEntryList);
putRecordsResult = amazonKinesisClient.putRecords(putRecordsRequest);
}
```
### Ajoutez un seul enregistrement avec PutRecord
Chaque appel de [https://docs.aws.amazon.com/kinesis/latest/APIReference/API_PutRecord.html](https://docs.aws.amazon.com/kinesis/latest/APIReference/API_PutRecord.html) est exécuté sur un seul enregistrement. Préférez l'opération `PutRecords` décrite dans [Ajoutez plusieurs enregistrements avec PutRecords](#kinesis-using-sdk-java-putrecords), sauf si votre application a particulièrement besoin de toujours envoyer des enregistrements uniques par demande ou qu'une autre raison empêche l'utilisation de `PutRecords`.
Chaque enregistrement de données a un numéro de séquence unique. Le numéro de séquence est attribué par Kinesis Data Streams une fois que vous avez appelé `client.putRecord` pour ajouter l'enregistrement de données au flux. Les numéros de séquence correspondant à une même clé de partition deviennent généralement de plus en plus longs au fil du temps ; plus l'intervalle de temps entre chaque demande `PutRecord` est élevé, plus les numéros de séquence sont longs.
Lorsque des opérations se succèdent rapidement, il n'est pas sûr que les numéros de séquence renvoyés augmentent, car ces opérations semblent surtout simultanées pour Kinesis Data Streams. Pour garantir une stricte augmentation des numéros de séquence pour la même clé de partition, utilisez le paramètre `SequenceNumberForOrdering`, comme il est illustré dans l'exemple de code [PutRecord exemple](#kinesis-using-sdk-java-putrecord-example).
Que vous utilisiez `SequenceNumberForOrdering` ou non, les enregistrements que Kinesis Data Streams reçoit via un appel `GetRecords` sont strictement classés par numéro de séquence.
**Note**
Les numéros de séquence ne peuvent pas servir d'index aux ensembles de données d'un même flux. Pour séparer logiquement les ensembles de données, utilisez des clés de partition ou créez un flux distinct pour chaque ensemble de données.
La clé de partition sert à regrouper les données dans le flux. Un enregistrement de données est attribué à une partition du flux suivant sa clé de partition. Plus précisément, Kinesis Data Streams utilise la clé de partition comme entrée d'une fonction de hachage qui mappe la clé de partition (et les données associées) à une partition spécifique.
Ce mécanisme de hachage a pour effet que tous les enregistrements de données ayant la même clé de partition sont mappés à la même partition du flux. Toutefois, si le nombre de clés de partition dépasse le nombre de partitions, certaines partitions contiennent nécessairement des enregistrements ayant des clés de partition différentes. Du point de vue de la conception, afin de garantir que toutes vos partitions sont bien utilisées, le nombre de partitions (spécifié par la méthode `setShardCount` de `CreateStreamRequest`) doit être nettement inférieur à celui des partitions uniques, et la quantité de données qui passe dans une clé de partition unique doit être nettement inférieure à la capacité de la partition.
#### PutRecord exemple
Le code suivant crée dix enregistrements de données répartis entre deux clés de partition et les place dans un flux appelé `myStreamName`.
```
for (int j = 0; j < 10; j++)
{
PutRecordRequest putRecordRequest = new PutRecordRequest();
putRecordRequest.setStreamName( myStreamName );
putRecordRequest.setData(ByteBuffer.wrap( String.format( "testData-%d", j ).getBytes() ));
putRecordRequest.setPartitionKey( String.format( "partitionKey-%d", j/5 ));
putRecordRequest.setSequenceNumberForOrdering( sequenceNumberOfPreviousRecord );
PutRecordResult putRecordResult = client.putRecord( putRecordRequest );
sequenceNumberOfPreviousRecord = putRecordResult.getSequenceNumber();
}
```
L'exemple de code précédent utilise le paramètre `setSequenceNumberForOrdering` pour garantir un ordre croissant dans chaque clé de partition. Pour utiliser ce paramètre efficacement, définissez le `SequenceNumberForOrdering` de l'enregistrement en cours (enregistrement *n*) sur le numéro de séquence de l'enregistrement précédent (enregistrement *n-1*). Pour obtenir le numéro de séquence d'un enregistrement qui a été ajouté au flux, appelez `getSequenceNumber` sur le résultat de `putRecord`.
Le paramètre `SequenceNumberForOrdering` garantit des numéros de séquence strictement croissants pour la même clé de partition. `SequenceNumberForOrdering` ne propose pas l'ordre des enregistrements sur plusieurs clés de partition.
# Interagissez avec les données à l'aide du registre des AWS Glue schémas
Vous pouvez intégrer vos flux de données Kinesis au registre des AWS Glue schémas. Le registre des AWS Glue schémas vous permet de découvrir, de contrôler et de faire évoluer les schémas de manière centralisée, tout en garantissant que les données produites sont validées en permanence par un schéma enregistré. Un schéma définit la structure et le format d'un enregistrement de données. Un schéma est une spécification versionnée pour la publication, la consommation ou le stockage des données fiables. Le registre des AWS Glue schémas vous permet d'améliorer la qualité end-to-end des données et la gouvernance des données au sein de vos applications de streaming. Pour plus d'informations, consultez le registre [AWS Glue Schema](https://docs.aws.amazon.com/glue/latest/dg/schema-registry.html) (français non garanti). L'un des moyens de configurer cette intégration consiste à utiliser le `PutRecords` kit `PutRecord` Kinesis Data APIs Streams disponible dans AWS le SDK Java.
Pour obtenir des instructions détaillées sur la façon de configurer l'intégration de Kinesis Data Streams au registre des schémas à l'aide PutRecords de Kinesis Data Streams PutRecord et de Kinesis Data APIs Streams, consultez la [section « Interaction avec les données à l'aide des Kinesis APIs Data Streams » dans Use Case : Integrating Amazon Kinesis Data Streams](https://docs.aws.amazon.com/glue/latest/dg/schema-registry-integrations.html#schema-registry-integrations-kds) with the Glue Schema Registry. AWS
# Écrire sur Amazon Kinesis Data Streams à l'aide de Kinesis Agent
L'agent Kinesis est une application logicielle Java autonome qui permet de collecter et d'envoyer facilement des données à Kinesis Data Streams. L'agent surveille en permanence un ensemble de fichiers et envoie de nouvelles données à votre flux. L'agent gère les rotations de fichier, les points de contrôle et les nouvelles tentatives après échec. Il diffuse toutes vos données de manière fiable, rapide et simple. Il émet également des CloudWatch métriques Amazon pour vous aider à mieux surveiller et résoudre les problèmes liés au processus de streaming.
Par défaut, les enregistrements sont analysés à partir de chaque fichier sur la base du caractère de saut de ligne (`'\n'`). Toutefois, l'agent peut également être configuré pour analyser les enregistrements de plusieurs lignes (consultez [Spécifiez les paramètres de configuration de l'agent](#agent-config-settings)).
Cet agent peut être installé dans des environnements basés sur des serveurs Linux tels que des serveurs Web, serveurs de journaux ou encore serveurs de base de données. Après avoir installé l'agent, configurez-le en spécifiant les fichiers à surveiller et le flux pour les données. Une fois que l'agent est configuré, il collecte de façon durable les données depuis les fichiers et les envoie en toute fiabilité dans le flux.
**Topics**
+ [
## Remplissez les conditions requises pour Kinesis Agent
](#prereqs)
+ [
## Télécharger et installer l’agent
](#download-install)
+ [
## Configuration et démarrage de l'agent
](#config-start)
+ [
## Spécifiez les paramètres de configuration de l'agent
](#agent-config-settings)
+ [
## Surveillez plusieurs répertoires de fichiers et écrivez dans plusieurs flux
](#sim-writes)
+ [
## Utiliser l'agent pour prétraiter les données
](#pre-processing)
+ [
## Utiliser les commandes CLI de l'agent
](#cli-commands)
+ [
## FAQ
](#agent-faq)
## Remplissez les conditions requises pour Kinesis Agent
+ Votre système d'exploitation doit être soit Amazon Linux AMI en version 2015.09 ou ultérieure, soit Red Hat Enterprise Linux en version 7 ou ultérieure.
+ Si vous utilisez Amazon EC2 pour exécuter votre agent, lancez votre instance EC2.
+ Gérez vos AWS informations d'identification à l'aide de l'une des méthodes suivantes :
+ Spécifiez un rôle IAM lorsque vous lancez votre instance EC2.
+ Spécifiez AWS les informations d'identification lorsque vous configurez l'agent (voir [awsAccessKeyID](#awsAccessKeyId) et [awsSecretAccessclé](#awsSecretAccessKey)).
+ Modifiez `/etc/sysconfig/aws-kinesis-agent` pour spécifier votre région et vos clés AWS d'accès.
+ [[Si votre instance EC2 se trouve sur un autre AWS compte, créez un rôle IAM pour fournir un accès au service Kinesis Data Streams, et spécifiez ce rôle lorsque vous configurez l'agent (voir AssumeRoLearn et Id). assumeRoleExternal](#assumeRoleExternalId)](#assumeRoleARN) Utilisez l'une des méthodes précédentes pour spécifier les AWS informations d'identification d'un utilisateur de l'autre compte autorisé à assumer ce rôle.
+ Le rôle ou les AWS informations d'identification IAM que vous spécifiez doivent être autorisés à effectuer l'opération Kinesis Data [PutRecords](https://docs.aws.amazon.com/kinesis/latest/APIReference/API_PutRecords.html)Streams pour que l'agent envoie des données à votre flux. Si vous activez la CloudWatch surveillance pour l'agent, l'autorisation d'effectuer l' CloudWatch [PutMetricData](https://docs.aws.amazon.com/AmazonCloudWatch/latest/APIReference/API_PutMetricData.html)opération est également nécessaire. Pour plus d'informations, consultez [Contrôle de l'accès aux ressources Amazon Kinesis Data Streams à l'aide d'IAM](controlling-access.md)[Surveillez l'état de santé des agents Kinesis Data Streams avec Amazon CloudWatch](agent-health.md), et [Contrôle CloudWatch d'accès](https://docs.aws.amazon.com/AmazonCloudWatch/latest/DeveloperGuide/UsingIAM.html).
## Télécharger et installer l’agent
Commencez par vous connecter à votre instance. Pour plus d'informations, consultez [Connect to Your Instance](https://docs.aws.amazon.com/AWSEC2/latest/UserGuide/ec2-connect-to-instance-linux.html) dans le guide de l'*utilisateur Amazon EC2*. Si vous rencontrez des difficultés pour vous connecter, consultez la section [Résolution des problèmes liés à la connexion à votre instance](https://docs.aws.amazon.com/AWSEC2/latest/UserGuide/TroubleshootingInstancesConnecting.html) dans le *guide de l'utilisateur Amazon EC2*.
**Pour configurer l'agent à l'aide d'Amazon Linux AMI**
Utilisez la commande suivante pour télécharger et installer l'agent :
```
sudo yum install –y aws-kinesis-agent
```
**Pour configurer l'agent à l'aide de Red Hat Enterprise Linux**
Utilisez la commande suivante pour télécharger et installer l'agent :
```
sudo yum install –y https://s3.amazonaws.com/streaming-data-agent/aws-kinesis-agent-latest.amzn2.noarch.rpm
```
**Pour configurer l'agent à l'aide de GitHub**
1. Téléchargez l'agent depuis [awlabs/ amazon-kinesis-agent](https://github.com/awslabs/amazon-kinesis-agent).
1. Installez l'agent en accédant au répertoire de téléchargement et en exécutant la commande suivante :
```
sudo ./setup --install
```
**Configuration de l'agent dans un conteneur Docker**
L'agent Kinesis peut également être exécuté dans un conteneur via la base de conteneurs [amazonlinux](https://docs.aws.amazon.com/AmazonECR/latest/userguide/amazon_linux_container_image.html). Utilisez le Dockerfile suivant, puis exécutez `docker build`.
```
FROM amazonlinux
RUN yum install -y aws-kinesis-agent which findutils
COPY agent.json /etc/aws-kinesis/agent.json
CMD ["start-aws-kinesis-agent"]
```
## Configuration et démarrage de l'agent
**Pour configurer et démarrer l'agent**
1. Ouvrez le fichier de configuration et modifiez-le (en tant que super-utilisateur si vous utilisez les autorisations d'accès fichier par défaut) :`/etc/aws-kinesis/agent.json`
Dans ce fichier de configuration, spécifiez les fichiers ( `"filePattern"` ) à partir desquels l'agent collecte les données, et le nom du flux ( `"kinesisStream"` ) dans lequel l'agent envoie les données. Notez que le nom de fichier est un modèle et que l'agent reconnaît les rotations de fichier. Vous pouvez effectuer une rotation de fichier ou créer de nouveaux fichiers pas plus d'une fois par seconde. L'agent utilise l'horodatage de création de fichier pour déterminer quels fichiers doivent être suivis et tracés dans votre flux. Créer de nouveaux fichiers ou effectuer une rotation de fichier plus souvent qu'une fois par seconde ne permet pas à l'agent de différencier correctement les fichiers.
```
{
"flows": [
{
"filePattern": "/tmp/app.log*",
"kinesisStream": "yourkinesisstream"
}
]
}
```
1. Démarrez l'agent manuellement :
```
sudo service aws-kinesis-agent start
```
1. (Facultatif) Configurez l'agent pour qu'il soit lancé au démarrage du système :
```
sudo chkconfig aws-kinesis-agent on
```
L'agent fonctionne maintenant en arrière-plan en tant que service système. Il surveille en permanence les fichiers spécifiés et envoie des données dans le flux spécifié. L'activité de l'agent est enregistrée dans `/var/log/aws-kinesis-agent/aws-kinesis-agent.log`.
## Spécifiez les paramètres de configuration de l'agent
L'agent prend en charge les deux paramètres de configuration obligatoires `filePattern` et `kinesisStream`, plus les paramètres de configuration facultatifs des fonctionnalités supplémentaires. Vous pouvez spécifier aussi bien une configuration obligatoire que facultative dans `/etc/aws-kinesis/agent.json`.
Chaque fois que vous modifiez le fichier de configuration, vous devez arrêter et démarrer l'agent en utilisant les commandes suivantes :
```
sudo service aws-kinesis-agent stop
sudo service aws-kinesis-agent start
```
Vous pouvez également utiliser la commande suivante :
```
sudo service aws-kinesis-agent restart
```
Les paramètres de configuration générale sont indiqués ci-après.
| Paramètre de configuration | Description |
| --- | --- |
| assumeRoleARN | ARN du rôle à être assumé par l'utilisateur. Pour plus d'informations, consultez la section [Accès délégué entre AWS comptes à l'aide de rôles IAM](https://docs.aws.amazon.com/IAM/latest/UserGuide/tutorial_cross-account-with-roles.html) dans le guide de l'*utilisateur IAM*. |
| assumeRoleExternalId | Identifiant facultatif qui détermine qui peut assumer le rôle. Pour en savoir plus, consultez la rubrique [Procédure d'utilisation d'un ID externe](https://docs.aws.amazon.com/IAM/latest/UserGuide/id_roles_create_for-user_externalid.html) dans le *Guide de l'utilisateur IAM*. |
| awsAccessKeyId | AWS ID de clé d'accès qui remplace les informations d'identification par défaut. Ce paramètre est prioritaire sur tous les autres fournisseurs d'informations d'identification. |
| awsSecretAccessKey | AWS clé secrète qui remplace les informations d'identification par défaut. Ce paramètre est prioritaire sur tous les autres fournisseurs d'informations d'identification. |
| cloudwatch.emitMetrics | Permet à l'agent d'émettre des métriques CloudWatch s'il est défini (true). Valeur par défaut : true |
| cloudwatch.endpoint | Le point de terminaison régional pour CloudWatch. Valeur par défaut : `monitoring.us-east-1.amazonaws.com` |
| kinesis.endpoint | Point de terminaison régional pour Kinesis Data Streams. Valeur par défaut : `kinesis.us-east-1.amazonaws.com` |
Les paramètres de configuration de flux sont indiqués ci-après.
| Paramètre de configuration | Description |
| --- | --- |
| dataProcessingOptions | Liste des options de traitement appliquées à chaque enregistrement analysé avant qu'il ne soit envoyé au flux. Les options de traitement sont exécutées dans l'ordre spécifié. Pour de plus amples informations, veuillez consulter [Utiliser l'agent pour prétraiter les données](#pre-processing). |
| kinesisStream | [Obligatoire] Nom du flux. |
| filePattern | [Obligatoire] Le répertoire et le modèle de fichier qui doivent correspondre pour être récupérés par l'agent. Pour tous les fichiers correspondant à ce modèle, l'autorisation en lecture doit être accordée à `aws-kinesis-agent-user`. Pour le répertoire contenant les fichiers, les autorisations en lecture et exécution doivent être accordées à `aws-kinesis-agent-user`. |
| initialPosition | Position initiale à partir de laquelle le fichier a commencé à être analysé. Les valeurs valides sont `START_OF_FILE` et `END_OF_FILE`. Valeur par défaut : `END_OF_FILE` |
| maxBufferAgeMillis | Durée maximale, en millisecondes, pendant laquelle l'agent met les données en tampon avant de les envoyer dans le flux. Plage de valeurs : 1 000 à 900 000 (1 seconde à 15 minutes) Par défaut : 60 000 (1 minute) |
| maxBufferSizeBytes | Taille maximale, en octets, pour laquelle l'agent met les données en tampon avant de les envoyer dans le flux. Plage de valeurs : 1 à 4 194 304 (4 Mo) Par défaut : 4 194 304 (4 Mo) |
| maxBufferSizeRecords | Nombre maximal d'enregistrements pour lequel l'agent met les données en tampon avant de les envoyer dans le flux. Plage de valeurs : 1 à 500 Par défaut : 500 |
| minTimeBetweenFilePollsMillis | Fréquence, en millisecondes, à laquelle l'agent interroge et analyse les fichiers surveillés pour rechercher les nouvelles données. Plage de valeurs : 1 ou plus Par défaut : 100 |
| multiLineStartPattern | Modèle d'identification du début d'un enregistrement. Un enregistrement se compose d'une ligne qui correspond au modèle et de lignes suivantes qui ne correspondent pas au modèle. Les valeurs valides sont les expressions régulières. Par défaut, chaque nouvelle ligne comprise dans les fichiers journaux est analysée comme étant un enregistrement. |
| partitionKeyOption | Méthode de génération de la clé de partition. Les valeurs valides sont `RANDOM` (entier généré uniquement aléatoirement) et `DETERMINISTIC` (une valeur de hachage calculée à partir des données). Valeur par défaut : `RANDOM` |
| skipHeaderLines | Nombre de lignes que l'agent doit ignorer lors de l'analyse au début des fichiers surveillés. Plage de valeurs : 0 ou plus Par défaut : 0 (zéro) |
| truncatedRecordTerminator | Chaîne que l'agent utilise pour tronquer un enregistrement analysé lorsque la taille de ce dernier dépasse la taille limite d'un enregistrement Kinesis Data Streams. (1,000 Ko) Par défaut : `'\n'` (nouvelle ligne) |
## Surveillez plusieurs répertoires de fichiers et écrivez dans plusieurs flux
En spécifiant plusieurs paramètres de configuration de flux, vous pouvez configurer l'agent pour surveiller plusieurs répertoires de fichiers et envoyer des données dans plusieurs flux. Dans l'exemple de configuration suivant, l'agent surveille deux répertoires de fichiers et envoie des données respectivement à un flux Kinesis et à un flux de diffusion Firehose. Notez que vous pouvez spécifier différents points de terminaison pour Kinesis Data Streams et Firehose afin que votre flux Kinesis et votre flux de diffusion Firehose ne soient pas nécessairement situés dans la même région.
```
{
"cloudwatch.emitMetrics": true,
"kinesis.endpoint": "https://your/kinesis/endpoint",
"firehose.endpoint": "https://your/firehose/endpoint",
"flows": [
{
"filePattern": "/tmp/app1.log*",
"kinesisStream": "yourkinesisstream"
},
{
"filePattern": "/tmp/app2.log*",
"deliveryStream": "yourfirehosedeliverystream"
}
]
}
```
Pour des informations plus détaillées sur l'utilisation de l'agent avec Firehose, consultez [Writing to Amazon Data Firehose with](https://docs.aws.amazon.com/firehose/latest/dev/writing-with-agents.html) Kinesis Agent.
## Utiliser l'agent pour prétraiter les données
L'agent peut prétraiter les enregistrements analysés à partir des fichiers surveillés avant de les envoyer dans votre flux. Vous pouvez activer cette fonctionnalité en ajoutant le paramètre de configuration `dataProcessingOptions` à votre flux de fichiers. Une ou plusieurs options de traitement peuvent être ajoutées. Elles sont exécutées dans l'ordre spécifié.
L'agent prend en charge les options de traitement répertoriées ci-après. Etant donné que l'agent est open source, vous pouvez continuer à développer et étendre ses options de traitement. Vous pouvez télécharger l'agent à partit de [Kinesis Agent](https://github.com/awslabs/amazon-kinesis-agent).Options de traitement
`SINGLELINE`
Convertit un enregistrement de plusieurs lignes en un enregistrement d'une seule ligne en supprimant les caractères de saut de ligne, les espaces de début et les espaces de fin.
```
{
"optionName": "SINGLELINE"
}
```
`CSVTOJSON`
Convertit un enregistrement du format séparé par délimiteur au format JSON.
```
{
"optionName": "CSVTOJSON",
"customFieldNames": [ "field1", "field2", ... ],
"delimiter": "yourdelimiter"
}
```
`customFieldNames`
[Obligatoire] Noms de domaine utilisés comme clés dans chaque paire clé-valeur JSON. Par exemple, si vous spécifiez `["f1", "f2"]`, l'enregistrement « v1, v2 » est converti en `{"f1":"v1","f2":"v2"}`.
`delimiter`
Chaîne utilisée comme délimiteur dans l'enregistrement. La valeur par défaut est une virgule (,).
`LOGTOJSON`
Convertit un enregistrement du format de journal au format JSON. Les formats de journal pris en charge sont **Apache Common Log**, **Apache Combined Log**, **Apache Error Log** et **RFC3164 Syslog**.
```
{
"optionName": "LOGTOJSON",
"logFormat": "logformat",
"matchPattern": "yourregexpattern",
"customFieldNames": [ "field1", "field2", … ]
}
```
`logFormat`
[Obligatoire] Format d'entrée de journal. Les valeurs admises sont les suivantes :
+ `COMMONAPACHELOG` : le format de journal courant d'Apache. Chaque entrée de journal a le schéma suivant par défaut : « `%{host} %{ident} %{authuser} [%{datetime}] \"%{request}\" %{response} %{bytes}` ».
+ `COMBINEDAPACHELOG` : le format de journal combiné d'Apache. Chaque entrée de journal a le schéma suivant par défaut : « `%{host} %{ident} %{authuser} [%{datetime}] \"%{request}\" %{response} %{bytes} %{referrer} %{agent}` ».
+ `APACHEERRORLOG` : le format de journal d'erreurs d'Apache. Chaque entrée de journal a le schéma suivant par défaut : « `[%{timestamp}] [%{module}:%{severity}] [pid %{processid}:tid %{threadid}] [client: %{client}] %{message}` ».
+ `SYSLOG`— Le format RFC3164 Syslog. Chaque entrée de journal a le schéma suivant par défaut : « `%{timestamp} %{hostname} %{program}[%{processid}]: %{message}` ».
`matchPattern`
Modèle d'expression régulière utilisé pour extraire les valeurs des entrées de journal. Ce paramètre est utilisé si votre entrée de journal n'a pas l'un des formats de journalisation prédéfinis. Si ce paramètre est utilisé, vous devez également spécifier `customFieldNames`.
`customFieldNames`
Noms de champ personnalisés utilisés comme clés dans chaque paire clé-valeur JSON. Vous pouvez utiliser ce paramètre pour définir les noms de champ pour les valeurs extraites de `matchPattern`, ou remplacer les noms de champ par défaut des formats de journalisation prédéfinis.
**Example : Configuration LOGTOJSON**
Voici un exemple de configuration `LOGTOJSON` pour une entrée au format Journal courant Apache convertie au format JSON :
```
{
"optionName": "LOGTOJSON",
"logFormat": "COMMONAPACHELOG"
}
```
Avant la conversion :
```
64.242.88.10 - - [07/Mar/2004:16:10:02 -0800] "GET /mailman/listinfo/hsdivision HTTP/1.1" 200 6291
```
Après la conversion :
```
{"host":"64.242.88.10","ident":null,"authuser":null,"datetime":"07/Mar/2004:16:10:02 -0800","request":"GET /mailman/listinfo/hsdivision HTTP/1.1","response":"200","bytes":"6291"}
```
**Example : Configuration LOGTOJSON avec champs personnalisés**
Voici un autre exemple de configuration `LOGTOJSON` :
```
{
"optionName": "LOGTOJSON",
"logFormat": "COMMONAPACHELOG",
"customFieldNames": ["f1", "f2", "f3", "f4", "f5", "f6", "f7"]
}
```
Avec ce paramètre de configuration, la même entrée au format Journal courant Apache de journal issue de l'exemple précédent est convertie au format JSON comme suit :
```
{"f1":"64.242.88.10","f2":null,"f3":null,"f4":"07/Mar/2004:16:10:02 -0800","f5":"GET /mailman/listinfo/hsdivision HTTP/1.1","f6":"200","f7":"6291"}
```
**Example : Conversion d'une entrée au format Journal courant Apache**
La configuration de flux suivante convertit une entrée au format Journal courant Apache en un enregistrement d'une seule ligne au format JSON :
```
{
"flows": [
{
"filePattern": "/tmp/app.log*",
"kinesisStream": "my-stream",
"dataProcessingOptions": [
{
"optionName": "LOGTOJSON",
"logFormat": "COMMONAPACHELOG"
}
]
}
]
}
```
**Example : Conversion des enregistrements de plusieurs lignes**
La configuration de flux suivante analyse les enregistrements de plusieurs lignes dont la première ligne commence par « `[SEQUENCE=` ». Chaque enregistrement est d'abord converti en un enregistrement d'une seule ligne. Les valeurs sont ensuite extraites de l'enregistrement sur la base d'un séparateur tabulation. Les valeurs extraites sont mappées à des valeurs `customFieldNames` spécifiées pour former un enregistrement d'une seule ligne au format JSON.
```
{
"flows": [
{
"filePattern": "/tmp/app.log*",
"kinesisStream": "my-stream",
"multiLineStartPattern": "\\[SEQUENCE=",
"dataProcessingOptions": [
{
"optionName": "SINGLELINE"
},
{
"optionName": "CSVTOJSON",
"customFieldNames": [ "field1", "field2", "field3" ],
"delimiter": "\\t"
}
]
}
]
}
```
**Example : Configuration LOGTOJSON avec modèle de correspondance**
Voici un exemple de configuration `LOGTOJSON` pour une entrée au format Journal courant Apache convertie au format JSON, avec le dernier champ (octets) omis :
```
{
"optionName": "LOGTOJSON",
"logFormat": "COMMONAPACHELOG",
"matchPattern": "^([\\d.]+) (\\S+) (\\S+) \\[([\\w:/]+\\s[+\\-]\\d{4})\\] \"(.+?)\" (\\d{3})",
"customFieldNames": ["host", "ident", "authuser", "datetime", "request", "response"]
}
```
Avant la conversion :
```
123.45.67.89 - - [27/Oct/2000:09:27:09 -0400] "GET /java/javaResources.html HTTP/1.0" 200
```
Après la conversion :
```
{"host":"123.45.67.89","ident":null,"authuser":null,"datetime":"27/Oct/2000:09:27:09 -0400","request":"GET /java/javaResources.html HTTP/1.0","response":"200"}
```
## Utiliser les commandes CLI de l'agent
Lancer automatiquement l'agent au démarrage du système :
```
sudo chkconfig aws-kinesis-agent on
```
Vérifier le statut de l'agent :
```
sudo service aws-kinesis-agent status
```
Arrêter l'agent :
```
sudo service aws-kinesis-agent stop
```
Lire le fichier journal de l'agent à partir de cet emplacement :
```
/var/log/aws-kinesis-agent/aws-kinesis-agent.log
```
Désinstaller l'agent :
```
sudo yum remove aws-kinesis-agent
```
## FAQ
### Existe-t-il un Kinesis Agent pour Windows ?
[Kinesis Agent pour Windows](https://docs.aws.amazon.com/kinesis-agent-windows/latest/userguide/what-is-kinesis-agent-windows.html) est un logiciel différent de Kinesis Agent pour les plateformes Linux.
### Pourquoi le ralentissement and/or `RecordSendErrors` de Kinesis Agent augmente-t-il ?
Cela est généralement dû à une limitation de Kinesis. Vérifiez la `WriteProvisionedThroughputExceeded` métrique pour Kinesis Data Streams ou `ThrottledRecords` la métrique pour Firehose Delivery Streams. Toute augmentation de ces métriques par rapport à 0 indique que les limites de flux doivent être augmentées. Pour plus d'informations, consultez les rubriques [Limites des flux de données Kinesis](https://docs.aws.amazon.com/streams/latest/dev/service-sizes-and-limits.html) (français non garanti) et [Flux de diffusion Amazon Firehose](https://docs.aws.amazon.com/firehose/latest/dev/limits.html) (français non garanti).
Une fois que vous avez exclu la limitation, vérifiez si Kinesis Agent est configuré pour suivre un grand nombre de petits fichiers. Il y a un délai lorsque Kinesis Agent suit un nouveau fichier, c'est pourquoi Kinesis Agent doit suivre un petit nombre de fichiers plus volumineux. Essayez de regrouper vos fichiers journaux dans des fichiers plus volumineux.
### Pourquoi est-ce que je reçois des exceptions `java.lang.OutOfMemoryError` ?
Kinesis Agent ne dispose pas de suffisamment de mémoire pour gérer sa charge de travail actuelle. Essayez d'augmenter `JAVA_START_HEAP` et `JAVA_MAX_HEAP` dans `/usr/bin/start-aws-kinesis-agent` et de redémarrer l'agent.
### Pourquoi est-ce que je reçois des exceptions `IllegalStateException : connection pool shut down` ?
Kinesis Agent ne dispose pas de suffisamment de connexions pour gérer sa charge de travail actuelle. Essayez d'augmenter `maxConnections` et `maxSendingThreads` dans les paramètres de configuration de votre agent général à `/etc/aws-kinesis/agent.json`. La valeur par défaut de ces champs est 12 fois supérieure au nombre de processeurs d'exécution disponibles. Consultez [AgentConfiguration.java](https://github.com/awslabs/amazon-kinesis-agent/blob/master/src/com/amazon/kinesis/streaming/agent/config/AgentConfiguration.java) pour en savoir plus sur les paramètres de configuration avancés des agents.
### Comment puis-je déboguer un autre problème avec Kinesis Agent ?
Les journaux de niveau `DEBUG` peuvent être activés dans `/etc/aws-kinesis/log4j.xml`.
### Comment dois-je configurer Kinesis Agent ?
Plus la taille de `maxBufferSizeBytes` est petite, plus Kinesis Agent envoie fréquemment des données. Cela peut être une bonne chose, car cela réduit le délai de livraison des enregistrements, mais cela augmente également le nombre de requêtes par seconde adressées à Kinesis.
### Pourquoi Kinesis Agent envoie-t-il des enregistrements dupliqués ?
Cela se produit en raison d'une mauvaise configuration dans le suivi des fichiers. Assurez-vous que chaque `fileFlow’s filePattern` ne correspond qu'à un seul fichier. Cela peut également se produire si le mode `logrotate` utilisé est en mode `copytruncate`. Essayez de remplacer le mode par défaut ou de créer le mode pour éviter les doublons. Pour plus d'informations sur la gestion des enregistrements dupliqués, consultez la rubrique [Gestion des enregistrements dupliqués](https://docs.aws.amazon.com/streams/latest/dev/kinesis-record-processor-duplicates.html) (français non garanti).
# Écrivez sur Kinesis Data Streams à l'aide d'autres services AWS
Les AWS services suivants peuvent s'intégrer directement à Amazon Kinesis Data Streams pour écrire des données dans des flux de données Kinesis. Passez en revue les informations relatives à chaque service qui vous intéresse et reportez-vous aux références fournies.
**Topics**
+ [
# Écrivez dans Kinesis Data Streams à l'aide de AWS Amplify
](using-other-services-amplify.md)
+ [
# Écrire sur Kinesis Data Streams à l'aide d'Amazon Aurora
](using-other-services-aurora.md)
+ [
# Écrivez sur Kinesis Data Streams à l'aide d'Amazon CloudFront
](using-other-services-CloudFront.md)
+ [
# Écrivez sur Kinesis Data Streams à l'aide d'Amazon Logs CloudWatch
](using-other-services-cw-logs.md)
+ [
# Écrivez sur Kinesis Data Streams à l'aide d'Amazon Connect
](using-other-services-connect.md)
+ [
# Écrivez dans Kinesis Data Streams à l'aide de AWS Database Migration Service
](using-other-services-migration.md)
+ [
# Écrire sur Kinesis Data Streams à l'aide d'Amazon DynamoDB
](using-other-services-ddb.md)
+ [
# Écrivez sur Kinesis Data Streams à l'aide d'Amazon EventBridge
](using-other-services-eventbridges.md)
+ [
# Écrivez dans Kinesis Data Streams à l'aide de AWS IoT Core
](using-other-services-iot-core.md)
+ [
# Écrire sur Kinesis Data Streams à l'aide d'Amazon Relational Database Service
](using-other-services-rds.md)
+ [
# Écrivez sur Kinesis Data Streams à l'aide d'Amazon Pinpoint
](using-other-services-pinpoint.md)
+ [
# Écrire dans Kinesis Data Streams à l'aide de la base de données Amazon Quantum Ledger (Amazon QLDB)
](using-other-services-quantum-ledger.md)
# Écrivez dans Kinesis Data Streams à l'aide de AWS Amplify
Vous pouvez utiliser Amazon Kinesis Data Streams pour diffuser des données à partir de vos applications mobiles conçues avec AWS Amplify pour un traitement en temps réel. Vous pouvez ensuite créer des tableaux de bord en temps réel, capturer les exceptions et générer des alertes, formuler des recommandations et prendre d'autres décisions commerciales ou opérationnelles en temps réel. Vous pouvez également envoyer des données à d'autres services tels qu'Amazon Simple Storage Service, Amazon DynamoDB et Amazon Redshift.
Pour plus d'informations, consultez la rubrique [Utilisation d'Amazon Kinesis](https://docs.amplify.aws/react/build-a-backend/more-features/analytics/streaming-data/) (français non garanti) dans le *Guide du développeur AWS Amplify*.
# Écrire sur Kinesis Data Streams à l'aide d'Amazon Aurora
Vous pouvez utiliser Amazon Kinesis Data Streams pour surveiller les activités de vos clusters de bases de données Amazon Aurora. Grâce aux flux d'activités de base de données, votre cluster de bases de données Aurora pousse les activités vers un flux de données Amazon Kinesis Data Streams en temps réel. Vous pouvez ensuite créer des applications de gestion de la conformité qui utilisent ces activités, les contrôlent et génèrent des alertes. Vous pouvez également utiliser Amazon Firehose pour stocker les données.
Pour plus d'informations, consultez la rubrique [Flux d'activités de base de données](https://docs.aws.amazon.com/AmazonRDS/latest/AuroraUserGuide/DBActivityStreams.html) dans le *Guide du développeur Amazon Aurora* (français non garanti).
# Écrivez sur Kinesis Data Streams à l'aide d'Amazon CloudFront
Vous pouvez utiliser Amazon Kinesis Data Streams CloudFront avec des journaux en temps réel et obtenir des informations sur les demandes adressées à une distribution en temps réel. Vous pouvez ensuite créer votre propre [consommateur de flux de données Kinesis](https://docs.aws.amazon.com/streams/latest/dev/building-consumers.html) ou utiliser Amazon Data Firehose pour envoyer les données de journal à Amazon S3, Amazon Redshift, Amazon OpenSearch Service ou à un service de traitement des journaux tiers.
Pour plus d'informations, consultez la section [Journaux en temps réel](https://docs.aws.amazon.com/AmazonCloudFront/latest/DeveloperGuide/real-time-logs.html) du manuel *Amazon CloudFront Developer Guide*.
# Écrivez sur Kinesis Data Streams à l'aide d'Amazon Logs CloudWatch
Vous pouvez utiliser des CloudWatch abonnements pour accéder à un flux en temps réel des événements du journal depuis Amazon CloudWatch Logs et le transmettre à un flux de données Kinesis à des fins de traitement, d'analyse et de chargement vers d'autres systèmes.
Pour plus d'informations, consultez la section [Traitement en temps réel des données de journal avec les abonnements](https://docs.aws.amazon.com/AmazonCloudWatch/latest/logs/Subscriptions.html) dans le *guide de l'utilisateur Amazon CloudWatch Logs*.
# Écrivez sur Kinesis Data Streams à l'aide d'Amazon Connect
Vous pouvez utiliser les flux de données Kinesis Data Streams pour exporter les enregistrements de contacts et les événements des agents en temps réel à partir de votre instance Amazon Connect. Vous pouvez également activer le streaming de données à partir des profils clients Amazon Connect pour recevoir automatiquement des mises à jour d'un flux de données Kinesis concernant la création de nouveaux profils ou la modification de profils existants.
Vous pouvez ensuite créer des applications grand public pour traiter et analyser les données en temps réel. Par exemple, en utilisant les enregistrements de contacts et les données de profil client, vous pouvez conserver les données de vos systèmes sources, telles que CRMs les outils d'automatisation du marketing, up-to-date avec les informations les plus récentes. En utilisant les données relatives aux événements des agents, vous pouvez créer des tableaux de bord qui affichent les informations et les événements relatifs aux agents, et déclencher des notifications personnalisées en cas d'activités spécifiques des agents.
Pour plus d'informations, consultez les rubriques [Flux de données pour votre instance](https://docs.aws.amazon.com/connect/latest/adminguide/data-streaming.html), [Configuration de l'exportation en temps réel](https://docs.aws.amazon.com/connect/latest/adminguide/set-up-real-time-export.html) et [Flux d'événements de l'agent](https://docs.aws.amazon.com/connect/latest/adminguide/agent-event-streams.html) dans le *Guide de l'administrateur Amazon Connect* (français non garanti).
# Écrivez dans Kinesis Data Streams à l'aide de AWS Database Migration Service
Vous pouvez l'utiliser AWS Database Migration Service pour migrer des données vers un flux de données Kinesis. Vous pouvez ensuite créer des applications consommateur qui traitent les enregistrements de données en temps réel. Vous pouvez également facilement envoyer des données en aval vers d'autres services tels qu'Amazon Simple Storage Service, Amazon DynamoDB et Amazon Redshift
Pour plus d'informations, consultez la rubrique [Utilisation de Kinesis Data Streams](https://docs.aws.amazon.com/dms/latest/userguide/CHAP_Target.Kinesis.html) dans le *Guide de l'utilisateur AWS Database Migration Service *.
# Écrire sur Kinesis Data Streams à l'aide d'Amazon DynamoDB
Vous pouvez utiliser Amazon Kinesis Data Streams pour capturer les modifications apportées à Amazon DynamoDB. Kinesis Data Streams récupère les modifications au niveau élément dans n’importe quelle table DynamoDB et les réplique dans un flux de données Kinesis de votre choix. Vos applications consommateur peuvent accéder à ce flux pour visualiser les modifications au niveau de l'élément en temps réel et transmettre ces modifications en aval ou prendre des mesures en fonction du contenu.
Pour plus d'informations, consultez la rubrique [Fonctionnement des flux de données Kinesis Data Streams avec DynamoDB](https://docs.aws.amazon.com/amazondynamodb/latest/developerguide/kds.html) dans le *Guide du développeur Amazon DynamoDB*.
# Écrivez sur Kinesis Data Streams à l'aide d'Amazon EventBridge
Kinesis Data Streams vous permet d' AWS envoyer des événements d'[appel](https://docs.aws.amazon.com/eventbridge/latest/userguide/eb-events.html) d'API EventBridge dans un flux, de créer des applications grand public et de traiter de grandes quantités de données. Vous pouvez également utiliser Kinesis Data Streams comme cible EventBridge dans Pipes et diffuser des enregistrements sous forme de flux à partir de l'une des sources disponibles après un filtrage et un enrichissement facultatifs.
Pour plus d'informations, consultez [Envoyer des événements vers un flux Amazon Kinesis](https://docs.aws.amazon.com/eventbridge/latest/userguide/eb-relay-events-kinesis-stream.html) et [EventBridge Pipes](https://docs.aws.amazon.com/eventbridge/latest/userguide/eb-pipes.html) dans le guide de * EventBridge l'utilisateur Amazon*.
# Écrivez dans Kinesis Data Streams à l'aide de AWS IoT Core
Vous pouvez écrire des données en temps réel à partir de messages MQTT dans AWS IoT Core en utilisant des actions AWS IoT Rule. Vous pouvez ensuite créer des applications qui traitent les données, analysent leur contenu, génèrent des alertes et les transmettent à des applications d'analyse ou à d'autres services AWS .
Pour plus d'informations, consultez la rubrique [Kinesis Data Streams](https://docs.aws.amazon.com/iot/latest/developerguide/kinesis-rule-action.html) (français non garanti) dans le *Guide du développeur AWS * (français non garanti).
# Écrire sur Kinesis Data Streams à l'aide d'Amazon Relational Database Service
Vous pouvez utiliser Amazon Kinesis Data Streams pour surveiller les activités de vos instances Amazon RDS. À l'aide des flux d'activité de base de données, Amazon RDS transmet les activités vers un flux de données Kinesis en temps réel. Vous pouvez ensuite créer des applications de gestion de la conformité qui utilisent ces activités, les contrôlent et génèrent des alertes. Vous pouvez également utiliser Amazon Data Firehose pour stocker les données.
Pour plus d'informations, consultez la rubrique [Flux d'activités de base de données](https://docs.aws.amazon.com/AmazonRDS/latest/UserGuide/DBActivityStreams.html) dans le *Guide du développeur Amazon RDS* (français non garanti).
# Écrivez sur Kinesis Data Streams à l'aide d'Amazon Pinpoint
Vous pouvez configurer Amazon Pinpoint pour qu'il envoie des données d'événements à Amazon Kinesis Data Streams. Amazon Pinpoint peut envoyer des données d'événements pour les campagnes, les parcours et les e-mails et SMS transactionnels. Vous avez ensuite la possibilité d'ingérer ces données dans des applications d'analyse ou de créer vos propres applications consommateur qui agissent en fonction du contenu des événements.
Pour plus d'informations, consultez la rubrique [Événements en flux continu](https://docs.aws.amazon.com/pinpoint/latest/developerguide/event-streams.html) dans le *Guide du développeur Amazon Pinpoint*.
# Écrire dans Kinesis Data Streams à l'aide de la base de données Amazon Quantum Ledger (Amazon QLDB)
Vous pouvez créer un flux dans Amazon QLDB qui capture chaque révision de document enregistrée dans votre journal et transmet ces données à Amazon Kinesis Data Streams en temps réel. Le flux QLDB représente un flux ininterrompu de données allant du journal de votre registre vers une ressource de flux de données Kinesis. Vous pouvez ensuite utiliser la plateforme de flux continu Kinesis ou la bibliothèque client Kinesis Client Library pour consommer votre flux, traiter les enregistrements de données et analyser le contenu des données. Le flux QLDB écrit vos données dans les flux de données Kinesis Data Streams sous la forme de trois types d'enregistrements : `control`, `block summary` et `revision details`.
Pour plus d'informations, consultez la rubrique [Flux](https://docs.aws.amazon.com/qldb/latest/developerguide/streams.html) dans le *Guide du développeur Amazon QLDB* (français non garanti).
# Écrivez sur Kinesis Data Streams à l'aide d'intégrations tierces
Vous pouvez écrire des données dans Kinesis Data Streams à l'aide de l'une des options tierces suivantes qui s'intègrent à Kinesis Data Streams. Sélectionnez l'option sur laquelle vous souhaitez en savoir plus et recherchez des ressources et des liens vers la documentation pertinente.
**Topics**
+ [
# Apache Flink
](using-other-services-flink.md)
+ [
# Fluentd
](using-other-services-Fluentd.md)
+ [
# Debezium
](using-other-services-Debezium.md)
+ [
# Oracle GoldenGate
](using-other-services-Oracle-GoldenGate.md)
+ [
# Kafka Connect
](using-other-services-kafka-connect.md)
+ [
# Adobe Experience
](using-other-services-adobe.md)
+ [
# Striim
](using-other-services-Striim.md)
# Apache Flink
Apache Flink est un environnement et un moteur de traitement distribué pour les calculs avec état sur des flux de données illimités et limités. Pour plus d'informations sur l'écriture dans Kinesis Data Streams à partir d'Apache Flink, consultez la rubrique [Connecteur Amazon Kinesis Data Streams](https://nightlies.apache.org/flink/flink-docs-release-1.15/docs/connectors/datastream/kinesis/).
# Fluentd
Fluentd est un collecteur de données open source pour une couche de journalisation unifiée. Pour plus d'informations sur l'écriture dans Kinesis Data Streams à partir de Fluentd. Pour plus d'informations, consultez la rubrique [PTraitement des flux à l'aide de Kinesis](https://docs.fluentd.org/how-to-guides/kinesis-stream).
# Debezium
Debezium est une plateforme distribuée open source dédiée à la capture des modifications de données. Pour plus d'informations sur l'écriture dans Kinesis Data Streams à partir de Debezium, consultez la rubrique [Diffusion en continu de modifications de données MySQL vers Amazon Kinesis](https://debezium.io/blog/2018/08/30/streaming-mysql-data-changes-into-kinesis/) (français non garanti).
# Oracle GoldenGate
Oracle GoldenGate est un produit logiciel qui vous permet de répliquer, de filtrer et de transformer des données d'une base de données à une autre. Pour plus d'informations sur l'écriture vers Kinesis Data Streams depuis GoldenGate Oracle, [voir Réplication de données vers Kinesis Data](https://blogs.oracle.com/dataintegration/post/data-replication-to-aws-kinesis-data-stream-using-oracle-goldengate) Stream à l'aide d'Oracle. GoldenGate
# Kafka Connect
Kafka Connect est un outil permettant de diffuser des données de manière évolutive et fiable entre Apache Kafka et d'autres systèmes. Pour plus d'informations sur l'écriture de données à partir d'Apache Kafka vers Kinesis Data Streams, consultez la rubrique [Connecteur Kinesis Kafka](https://github.com/awslabs/kinesis-kafka-connector) (français non garanti).
# Adobe Experience
Adobe Experience Platform permet aux entreprises de centraliser et de standardiser les données des consommateurs depuis n'importe quel système. Elle applique ensuite la science des données et le machine learning pour améliorer considérablement la conception et la fourniture d'expériences riches et personnalisées. Pour en savoir plus sur l'écriture de données à partir d'Adobe Experience Platform vers Kinesis Data Streams, consultez la procédure de création d'une [connexion Amazon Kinesis](https://experienceleague.adobe.com/docs/experience-platform/destinations/catalog/cloud-storage/amazon-kinesis.html?lang=en).
# Striim
Striim est une plateforme complète en mémoire pour la collecte end-to-end, le filtrage, la transformation, l'enrichissement, l'agrégation, l'analyse et la diffusion de données en temps réel. Pour plus d'informations sur la façon d'écrire des données dans Kinesis Data Streams à partir de Striim, consultez la rubrique [Composant d'écriture Kinesis](https://www.striim.com/docs/en/kinesis-writer.html).
# Résoudre les problèmes liés aux producteurs d'Amazon Kinesis Data Streams
**Topics**
+ [
## Ma candidature de producteur s'écrit plus lentement que prévu
](#producer-writing-at-slower-rate)
+ [
## Je reçois une erreur d'autorisation de clé principale KMS non autorisée
](#unauthorized-kms-producer)
+ [
## Résoudre d'autres problèmes courants rencontrés par les producteurs
](#misc-troubleshooting-producer)
## Ma candidature de producteur s'écrit plus lentement que prévu
**Topics**
+ [
### Limites de service dépassées
](#service-limits-exceeded)
+ [
### Je souhaite optimiser mon producteur
](#producer-optimization)
+ [
### Mauvaise utilisation des `flushSync()` opérations
](#misuse-tag)
### Limites de service dépassées
Pour savoir si les limites de service sont dépassées, vérifiez si votre application producteur émet des exceptions de débit depuis le service, et validez quelles opérations d'API sont limitées. N'oubliez pas qu'il existe différentes limites basées sur l'appel, consultez [Quotas et limites](service-sizes-and-limits.md). Par exemple, en plus des nombres limite d'écritures et de lectures au niveau d'une partition, qui sont connus la plupart du temps, les nombres limite suivants sont applicables au niveau du flux :
+ [CreateStream](https://docs.aws.amazon.com/kinesis/latest/APIReference/API_CreateStream.html)
+ [DeleteStream](https://docs.aws.amazon.com/kinesis/latest/APIReference/API_DeleteStream.html)
+ [ListStreams](https://docs.aws.amazon.com/kinesis/latest/APIReference/API_ListStreams.html)
+ [GetShardIterator](https://docs.aws.amazon.com/kinesis/latest/APIReference/API_GetShardIterator.html)
+ [MergeShards](https://docs.aws.amazon.com/kinesis/latest/APIReference/API_MergeShards.html)
+ [DescribeStream](https://docs.aws.amazon.com/kinesis/latest/APIReference/API_DescribeStream.html)
+ [DescribeStreamSummary](https://docs.aws.amazon.com/kinesis/latest/APIReference/API_DescribeStreamSummary.html)
Les opérations `CreateStream`, `DeleteStream`, `ListStreams`, `GetShardIterator` et `MergeShards` sont limitées à 5 appels par seconde. L'opération `DescribeStream` est limitée à 10 appels par seconde. L'opération `DescribeStreamSummary` est limitée à 20 appels par seconde.
Si ces appels ne constituent pas le problème, assurez-vous que vous avez sélectionné une clé de partition qui permet de répartir également les opérations *put* entre toutes les partitions, et que vous n'avez pas de clé de partition spécifique qui se heurte aux limites de service alors que le reste ne le fait pas. Pour cela, vous devez mesurer le débit maximal et tenir compte du nombre de partitions du flux. Pour plus d'informations sur la gestion des flux, consultez la page [Création et gestion de flux de données Kinesis](working-with-streams.md).
**Astuce**
N'oubliez pas d'arrondir au kilo octet le plus proche pour le calcul de la limitation de débit lorsque vous utilisez l'opération sur enregistrement unique [PutRecord](https://docs.aws.amazon.com/kinesis/latest/APIReference/API_PutRecord.html), alors que l'opération sur plusieurs enregistrements [PutRecords](https://docs.aws.amazon.com/kinesis/latest/APIReference/API_PutRecords.html) arrondit à la somme cumulative des enregistrements dans chaque appel. Par exemple, une demande `PutRecords` contenant 600 enregistrements de 1,1 Ko n'est pas limitée.
### Je souhaite optimiser mon producteur
Avant de commencer à optimiser votre producteur, effectuez les tâches clés suivantes. Vous devez commencer par identifier votre débit maximal voulu en termes de taille d'enregistrement et de nombre d'enregistrements par seconde. Eliminez ensuite la capacité de flux comme facteur limitant ([Limites de service dépassées](#service-limits-exceeded)). Si vous avez éliminé la capacité du flux, utilisez les conseils de dépannage et instructions d'optimisation suivants pour les deux types courants d'applications producteur.
**Application producteur de grande capacité**
Une application producteur de grande capacité fonctionne généralement sur un serveur sur site ou une instance Amazon EC2. Les clients qui ont besoin d'un débit supérieur pour une application producteur de grande capacité s'occupent de la latence par enregistrement. Les stratégies pour gérer la latence sont les suivantes : si le client peut microloter/mettre en mémoire tampon des enregistrements, utilisez la [bibliothèque Amazon Kinesis Producer](https://docs.aws.amazon.com/kinesis/latest/dev/developing-producers-with-kpl.html) (dotée d'une logique d'agrégation avancée), l'opération multi-enregistrements ou agréger les enregistrements dans un fichier plus volumineux avant d'utiliser l'opération [PutRecords](https://docs.aws.amazon.com/kinesis/latest/APIReference/API_PutRecords.html)d'enregistrement unique. [PutRecord](https://docs.aws.amazon.com/kinesis/latest/APIReference/API_PutRecord.html) Si vous ne pouvez pas regrouper/mettre en mémoire tampon, utilisez plusieurs threads pour écrire simultanément dans le service Kinesis Data Streams. Les AWS SDK pour Java autres SDKs incluent des clients asynchrones qui peuvent le faire avec très peu de code.
**Application producteur de petite capacité**
Une application producteur de petite capacité est généralement une application mobile, un appareil IoT ou un client Web. S'il s'agit d'une application mobile, nous vous recommandons d'utiliser l'`PutRecords`appareil ou l'enregistreur Kinesis dans le AWS mobile. SDKs Pour plus d'informations, consultez le AWS Mobile SDK pour Android Guide de démarrage et Applications mobiles AWS SDK for iOS le Guide de démarrage. Les applications mobiles doivent gérer les connexions intermittentes de façon intrinsèque et ont besoin d'une sorte de placement par lots, par exemple `PutRecords`. Si vous ne pouvez pas regrouper pour une raison quelconque, consultez les informations ci-dessus relatives aux applications producteur de grande capacité. Si votre application producteur est un navigateur, le volume de données généré est généralement très faible. Cependant, vous placez les opérations *put* sur le chemin critique de l'application, ce que nous ne recommandons pas.
### Mauvaise utilisation des `flushSync()` opérations
`flushSync()`Une utilisation incorrecte peut avoir un impact significatif sur les performances d'écriture. L'`flushSync()`opération est conçue pour les scénarios d'arrêt afin de garantir que tous les enregistrements mis en mémoire tampon sont envoyés avant la fin de l'application KPL. Si vous avez implémenté cette opération après chaque opération d'écriture, elle peut ajouter une latence supplémentaire substantielle, environ 500 ms par écriture. Assurez-vous que vous n'avez implémenté `flushSync()` que pour l'arrêt de l'application afin d'éviter de retarder inutilement les performances d'écriture.
## Je reçois une erreur d'autorisation de clé principale KMS non autorisée
Cette erreur se produit lorsqu'une application producteur écrit sur un flux chiffré sans disposer d'autorisation sur la clé principale KMS. Pour attribuer à une application des autorisations d'accès à une clé KMS, consultez les rubriques [Utilisation des politiques de clés dans AWS KMS](https://docs.aws.amazon.com/kms/latest/developerguide/key-policies.html) et [Utilisation des politiques IAM avec AWS KMS](https://docs.aws.amazon.com/kms/latest/developerguide/iam-policies.html).
## Résoudre d'autres problèmes courants rencontrés par les producteurs
+ [Pourquoi mon flux de données Kinesis renvoie-t-il une erreur interne de serveur 500 ?](https://aws.amazon.com/premiumsupport/knowledge-center/kinesis-data-stream-500-error/)
+ [Comment résoudre les erreurs de dépassement de délai lors de l'écriture à partir de Flink vers Kinesis Data Streams ?](https://aws.amazon.com/premiumsupport/knowledge-center/kinesis-flink-timeout/)
+ [Comment résoudre les erreurs de limitation dans Kinesis Data Streams ?](https://aws.amazon.com/premiumsupport/knowledge-center/kinesis-data-stream-throttling-errors/)
+ [Pourquoi mon flux de données Kinesis est-il limité ?](https://aws.amazon.com/premiumsupport/knowledge-center/kinesis-data-stream-throttling/)
+ [Comment puis-je insérer des enregistrements de données dans un flux de données Kinesis à l'aide de KPL ?](https://aws.amazon.com/premiumsupport/knowledge-center/kinesis-data-stream-kpl/)
# Optimisez les producteurs de Kinesis Data Streams
Vous pouvez optimiser davantage vos producteurs Amazon Kinesis Data Streams en fonction du comportement spécifique que vous observez. Consultez les rubriques suivantes pour identifier des solutions.
**Topics**
+ [
# Personnalisation des nouvelles tentatives KPL et du comportement des limites de débit
](kinesis-producer-adv-retries-rate-limiting.md)
+ [
# Appliquer les meilleures pratiques à l'agrégation KPL
](kinesis-producer-adv-aggregation.md)
# Personnalisation des nouvelles tentatives KPL et du comportement des limites de débit
Lorsque vous ajoutez des enregistrements utilisateur Amazon Kinesis Producer Library (KPL) à l'aide de l'`addUserRecord()`opération KPL, un enregistrement reçoit un horodatage et est ajouté à une mémoire tampon avec une date limite définie par le paramètre de configuration. `RecordMaxBufferedTime` Cette stamp/deadline combinaison de temps définit la priorité de la mémoire tampon. Les enregistrements sont vidés de la mémoire tampon selon les critères suivants :
+ Priorité tampon
+ Configuration du regroupement
+ Configuration de la collecte
Les paramètres de configuration du regroupement et de la collecte qui influencent le comportement du tampon sont les suivants :
+ `AggregationMaxCount`
+ `AggregationMaxSize`
+ `CollectionMaxCount`
+ `CollectionMaxSize`
Les enregistrements vidés sont ensuite envoyés à votre flux de données Kinesis sous forme d'enregistrements Amazon Kinesis Data Streams à l'aide d'un appel à l'opération d'API Kinesis Data Streams `PutRecords`. L'opération `PutRecords` envoie des demandes à votre flux qui échouent parfois entièrement ou partiellement. Les enregistrements qui échouent sont automatiquement renvoyés au tampon KPL. Le nouveau délai est défini sur la base du minimum des deux valeurs suivantes :
+ La moitié de la configuration `RecordMaxBufferedTime` actuelle
+ La time-to-live valeur de l'enregistrement
Cette stratégie permet d'inclure les enregistrements utilisateur KPL réessayés dans les appels d'API Kinesis Data Streams suivants, afin d'améliorer le débit et de réduire la complexité tout en renforçant la valeur de l'enregistrement Kinesis Data Streams. time-to-live Il n'y a aucun algorithme d'interruption, ce qui rend cette stratégie de nouvelle tentative relativement brutale. Le spam provoqué par des tentatives excessives est empêché par la limitation de vitesse, qui est présentée à la section suivante.
## Limitation du débit
La KPL comprend une fonctionnalité de limitation de vitesse, qui limite le débit par partition envoyé par une seule application producteur. La limitation de vitesse est implémentée à l'aide d'un algorithme de compartiment à jeton, avec des compartiments distincts pour les enregistrements Kinesis Data Streams et les octets. Chaque écriture réussie dans un flux de données Kinesis ajoute un jeton (ou plusieurs jetons) à chaque compartiment, jusqu'à un certain seuil. Ce seuil est configurable, mais est défini par défaut sur une valeur 50 % supérieure au nombre limite de partitions réel afin d'autoriser la saturation de partition par une seule application producteur.
Vous pouvez réduire ce nombre limite pour diminuer le spam dû aux tentatives excessives. Toutefois, la bonne pratique consiste, pour chaque application producteur, à effectuer des tentatives brutales pour atteindre un débit maximal et à traiter toute limitation obtenue comme excessive en étendant la capacité du flux et en mettant en œuvre une stratégie de clé de partition appropriée.
# Appliquer les meilleures pratiques à l'agrégation KPL
Bien que le schéma de numérotation des enregistrements Amazon Kinesis Data Streams obtenus reste le même, l'agrégation entraîne l'indexation des enregistrements utilisateur Amazon Kinesis Producer Library (KPL) contenus dans un enregistrement Kinesis Data Streams agrégé à partir de 0 (zéro) ; toutefois, tant que vous ne vous fiez pas aux numéros de séquence pour identifier de manière unique vos enregistrements utilisateur KPL, votre code peut l'ignorer, car l'agrégation (de vos enregistrements utilisateur KPL dans un enregistrement Kinesis Data Streams) et désagrégation ultérieure (d'un enregistrement Kinesis Data Streams dans vos enregistrements utilisateur KPL) s'en charge automatiquement pour vous. Cela s'applique que votre consommateur utilise la KCL ou le AWS SDK. Pour utiliser cette fonctionnalité d'agrégation, vous devez intégrer la partie Java du KPL dans votre build si votre client est écrit à l'aide de l'API fournie dans le AWS SDK.
Si vous prévoyez d'utiliser les numéros de séquence comme des identifiants uniques pour vos enregistrements utilisateur KPL, nous vous recommandons d'utiliser les opérations `public int hashCode()` et `public boolean equals(Object obj)` conformes au contrat et fournies dans `Record` et `UserRecord` pour permettre la comparaison de vos enregistrements utilisateur KPL. En outre, si vous souhaitez examiner le numéro de sous-séquence de votre enregistrement utilisateur KPL, vous pouvez le convertir en instance `UserRecord` et extraire son numéro de sous-séquence.
Pour de plus amples informations, veuillez consulter [Mettre en œuvre la désagrégation des consommateurs](kinesis-kpl-consumer-deaggregation.md).