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.
# Utilisation des flux de capture des données modifiées (CDC) dans Amazon Keyspaces
La capture des données de modification (CDC) d'Amazon Keyspaces enregistre les événements de modification au niveau des lignes à partir d'une table Amazon Keyspaces en temps quasi réel.
Amazon Keyspaces CDC permet des cas d'utilisation axés sur des événements tels que l'IoT industriel et la détection des fraudes, ainsi que des cas d'utilisation liés au traitement des données tels que la recherche en texte intégral et l'archivage des données. Les événements de changement capturés par Amazon Keyspaces CDC dans les flux peuvent être utilisés par les applications en aval qui exécutent des fonctions critiques telles que l'analyse des données, la recherche de texte, la formation/inférence de machine learning et les sauvegardes continues des données à des fins d'archivage. Par exemple, vous pouvez transférer des données de flux vers des services AWS d'analyse et de stockage tels qu'Amazon OpenSearch Service, Amazon Redshift et Amazon S3 pour un traitement ultérieur.
Amazon Keyspaces CDC propose des enregistrements de modifications triés dans le temps et dédupliqués pour les tables, avec une mise à l'échelle automatique du débit de données et une durée de conservation allant jusqu'à 24 heures.
Les flux CDC d'Amazon Keyspaces sont totalement sans serveur et vous n'avez pas besoin de gérer l'infrastructure de données pour capturer les événements de changement. En outre, Amazon Keyspaces CDC ne consomme aucune capacité de table, que ce soit pour le calcul ou le stockage. Pour de plus amples informations, veuillez consulter [Comment fonctionnent les flux de capture des données modifiées (CDC) dans Amazon Keyspaces](cdc_how-it-works.md).
Vous pouvez utiliser l'API Amazon Keyspaces Streams pour créer des applications qui utilisent les flux CDC d'Amazon Keyspaces et prennent des mesures en fonction de leur contenu. Pour les points de terminaison disponibles, voir[Comment accéder aux points de terminaison CDC Stream dans Amazon Keyspaces](CDC_access-endpoints.md).
Pour une liste complète de toutes les opérations disponibles pour Amazon Keyspaces dans l'API Streams, consultez le manuel de référence de l'API Amazon [https://docs.aws.amazon.com/keyspaces/latest/StreamsAPIReference/Welcome.html](https://docs.aws.amazon.com/keyspaces/latest/StreamsAPIReference/Welcome.html) Streams.
**Topics**
+ [Comment fonctionnent les flux de capture des données modifiées (CDC) dans Amazon Keyspaces](cdc_how-it-works.md)
+ [Comment utiliser les flux de capture des données modifiées (CDC) dans Amazon Keyspaces](cdc_how-to-use.md)
# Comment fonctionnent les flux de capture des données modifiées (CDC) dans Amazon Keyspaces
Cette section fournit un aperçu du fonctionnement des flux de capture des données de modification (CDC) dans Amazon Keyspaces.
La capture des données de modification (CDC) d'Amazon Keyspaces enregistre une séquence ordonnée de modifications au niveau des lignes dans les tables Amazon Keyspaces et stocke ces informations dans un journal appelé *stream* pendant 24 heures maximum. Chaque modification au niveau de la ligne génère un nouvel enregistrement CDC qui contient les informations de la colonne clé primaire ainsi que les états « avant » et « après » de la ligne, y compris toutes les colonnes. Les applications peuvent accéder au flux et visualiser les mutations en temps quasi réel.
Lorsque vous activez le CDC sur votre table, Amazon Keyspaces crée un nouveau flux CDC et commence à recueillir des informations sur chaque modification de la table. Le flux CDC possède un Amazon Resource Name (ARN) au format suivant :
```
arn:${Partition}:cassandra:{Region}:${Account}:/keyspace/${keyspaceName}/table/${tableName}/stream/${streamLabel}
```
Vous pouvez sélectionner le type d'informations ou le *type de vue* que le flux CDC collecte pour chaque enregistrement lorsque vous activez le flux CDC pour la première fois. Vous ne pouvez pas modifier le type de vue du stream par la suite. Amazon Keyspaces prend en charge les types de vues suivants :
+ `NEW_AND_OLD_IMAGES`— Capture les versions de la ligne avant et après la mutation. Il s’agit de l’option par défaut.
+ `NEW_IMAGE`— Capture la version de la ligne après la mutation.
+ `OLD_IMAGE`— Capture la version de la ligne avant la mutation.
+ `KEYS_ONLY`— Capture les clés de partition et de clustering de la ligne qui a été mutée.
Chaque flux CDC est composé d'enregistrements. Chaque enregistrement représente une modification d'une seule ligne dans une table Amazon Keyspaces. Les enregistrements sont organisés de manière logique en groupes appelés *shards*. Ces groupes sont organisés logiquement par plages de clés primaires (combinaison de clés de partition, plages de clés de clustering) et constituent une construction interne d'Amazon Keyspaces. Chaque partition agit comme un conteneur pour plusieurs enregistrements et contient les informations nécessaires pour accéder à ces enregistrements et les itérer.
![\[Un flux CDC Amazon Keyspaces est constitué de fragments représentant un enregistrement CDC d'un ensemble de mutations de lignes.\]](http://docs.aws.amazon.com/fr_fr/keyspaces/latest/devguide/images/keyspaces_cdc.png)
Un numéro de séquence est attribué à chaque enregistrement CDC, qui reflète l'ordre dans lequel l'enregistrement a été publié dans le shard. Il est garanti que le numéro de séquence augmentera et sera unique au sein de chaque partition.
Amazon Keyspaces crée et supprime automatiquement les partitions. En fonction de la charge de trafic, Amazon Keyspaces peut également diviser ou fusionner des partitions au fil du temps. Par exemple, Amazon Keyspaces peut diviser une partition en plusieurs nouvelles partitions ou fusionner des partitions en une seule partition. Amazon Keyspaces APIs publie les informations du shard et du CDC pour permettre aux applications consommatrices de traiter les enregistrements dans le bon ordre en accédant au graphe de lignée complet d'un fragment.
Amazon Keyspaces CDC repose sur les principes suivants sur lesquels vous pouvez vous appuyer lors de la création de votre application :
+ Chaque enregistrement de mutation au niveau de la ligne apparaît exactement une fois dans le flux CDC.
+ Lorsque vous consommez des fragments par ordre de lignée, chaque enregistrement de mutation au niveau de la ligne apparaît dans la même séquence que l'ordre de mutation réel sur la clé primaire.
**Topics**
+ [Conservation des données](#CDC_how-it-works-data-retention)
+ [Expiration des données TTL](#CDC_how-it-works-ttl)
+ [Opérations par lots](#CDC_how-it-works-batch-operations)
+ [Colonnes statiques](#CDC_how-it-works-static)
+ [Chiffrement au repos](#CDC_how-it-works-encryption)
+ [Réplication multi-régions](#CDC_how-it-works-mrr)
+ [Intégration avec les AWS services](#howitworks_integration)
## Comment fonctionne la conservation des données pour les flux CDC dans Amazon Keyspaces
Amazon Keyspaces conserve les enregistrements du flux CDC pendant 24 heures. Vous ne pouvez pas modifier la durée de conservation. Si vous désactivez le CDC sur une table, les données du flux restent lisibles pendant 24 heures. Passé ce délai, les données expirent et les enregistrements sont automatiquement supprimés.
## Comment fonctionne l'expiration des données Time to Live (TTL) avec les flux CDC dans Amazon Keyspaces
Amazon Keyspaces indique le délai d'expiration au column/cell niveau ainsi qu'au niveau de la ligne dans un champ de métadonnées appelé `expirationTime` dans les enregistrements des modifications du CDC. Lorsque le TTL d'Amazon Keyspaces détecte l'expiration d'une cellule, le CDC crée un nouvel enregistrement de modification indiquant que le TTL est à l'origine de la modification. Pour plus d’informations sur TTL, consultez [Expirer les données avec Time to Live (TTL) pour Amazon Keyspaces (pour Apache Cassandra)](TTL.md).
## Comment fonctionnent les opérations par lots pour les flux CDC dans Amazon Keyspaces
Les opérations par lots sont divisées en interne en modifications individuelles au niveau des lignes. Amazon Keyspaces conserve tous les enregistrements des flux CDC au niveau de la ligne, même si la modification s'est produite lors d'une opération par lots. Amazon Keyspaces conserve l'ordre des enregistrements dans le flux CDC dans le même ordre que l'ordre de mutation qui s'est produit au niveau de la ligne ou sur la clé primaire.
## Comment fonctionnent les colonnes statiques dans les flux CDC sur Amazon Keyspaces
Les valeurs des colonnes statiques sont partagées entre toutes les lignes d'une partition dans Cassandra. En raison de ce comportement, Amazon Keyspaces capture toutes les mises à jour apportées à une colonne statique sous la forme d'un enregistrement distinct dans le flux CDC. Les exemples suivants résument le comportement des mutations de colonnes statiques :
+ Lorsque seule la colonne statique est mise à jour, le flux CDC contient une modification de ligne pour la colonne statique en tant que seule colonne de la ligne.
+ Lorsqu'une ligne est mise à jour sans aucune modification de la colonne statique, le flux CDC contient une modification de ligne qui contient toutes les colonnes sauf la colonne statique.
+ Lorsqu'une ligne est mise à jour en même temps que la colonne statique, le flux CDC contient deux modifications de ligne distinctes, l'une pour la colonne statique et l'autre pour le reste de la ligne.
## Comment fonctionne le chiffrement au repos pour les flux CDC dans Amazon Keyspaces
Pour chiffrer les données inactives dans le journal commandé par le CDC, Amazon Keyspaces utilise la même clé de chiffrement que celle déjà utilisée pour la table. Pour plus d’informations sur le chiffrement au repos, consultez [Le chiffrement est au repos dans Amazon Keyspaces](EncryptionAtRest.md).
## Comment fonctionne la réplication multirégionale pour les flux CDC dans Amazon Keyspaces
Vous pouvez activer et désactiver les flux CDC pour les répliques individuelles d'une table multirégionale à l'aide de l'`update-table`API ou de la commande `ALTER TABLE` CQL. En raison de la réplication asynchrone et de la résolution des conflits, les flux CDC pour les tables multirégionales ne sont pas cohérents entre eux. Régions AWS Par conséquent, les enregistrements capturés par Amazon Keyspaces dans le flux peuvent apparaître dans un ordre différent selon les régions.
Pour plus d'informations sur la réplication multirégionale, consultez[Réplication multirégionale pour Amazon Keyspaces (pour Apache Cassandra)](multiRegion-replication.md).
## Streams CDC et intégration aux AWS services
### Comment utiliser les points de terminaison VPC pour les flux CDC dans Amazon Keyspaces
Vous pouvez utiliser les points de terminaison VPC pour accéder aux flux CDC d'Amazon Keyspaces. Pour plus d'informations sur la création et l'accès aux points de terminaison VPC pour les flux, consultez. [Utilisation des flux CDC d'Amazon Keyspaces avec des points de terminaison VPC d'interface](vpc-endpoints-streams.md)
### Comment CloudWatch fonctionne la surveillance avec les flux CDC dans Amazon Keyspaces
Vous pouvez utiliser Amazon CloudWatch pour surveiller les appels d'API effectués vers le point de terminaison Amazon Keyspaces CDC. Pour plus d'informations sur les mesures disponibles, consultez[Mesures relatives à la capture des données de modification par Amazon Keyspaces (CDC)](metrics-dimensions.md#keyspaces-cdc-metrics).
### Comment CloudTrail fonctionne la journalisation avec les flux CDC sur Amazon Keyspaces
Amazon Keyspaces CDC est intégré à AWS CloudTrail un service qui fournit un enregistrement des actions effectuées par un utilisateur, un rôle ou un AWS service dans Amazon Keyspaces. CloudTrail capture les appels d'API DDL (Data Definition Language) et les appels d'API DML (Data Manipulation Language) pour Amazon Keyspaces sous forme d'événements. Les appels capturés incluent les appels provenant de la console Amazon Keyspaces et les appels programmatiques vers les opérations de l'API Amazon Keyspaces.
Pour plus d'informations sur les événements CDC capturés par CloudTrail, voir[Journalisation des appels d'API Amazon Keyspaces avec AWS CloudTrail](logging-using-cloudtrail.md).
### Comment fonctionne le balisage pour les flux CDC dans Amazon Keyspaces
Les flux CDC d'Amazon Keyspaces sont une ressource étiquetable. Vous pouvez baliser un flux lorsque vous créez une table par programmation à l'aide de CQL, du AWS SDK ou du. AWS CLI Vous pouvez également baliser des flux existants, supprimer des balises ou afficher les balises d'un flux. Pour de plus amples informations, veuillez consulter [Étiquetez les espaces clés, les tables et les flux dans Amazon Keyspaces](Tagging.Operations.md).
# Comment utiliser les flux de capture des données modifiées (CDC) dans Amazon Keyspaces
**Topics**
+ [Configurer les autorisations](configure-cdc-permissions.md)
+ [Accédez aux points de terminaison CDC Stream](CDC_access-endpoints.md)
+ [Activer un flux CDC pour une nouvelle table](keyspaces-enable-cdc-new-table.md)
+ [Activer un flux CDC pour une table existante](keyspaces-enable-cdc-alter-table.md)
+ [Désactiver un flux CDC](keyspaces-delete-cdc.md)
+ [Voir les streams du CDC](keyspaces-view-cdc.md)
+ [Accédez aux flux du CDC](keyspaces-records-cdc.md)
+ [Utiliser KCL pour traiter les flux](cdc_how-to-use-kcl.md)
# Configurer les autorisations pour utiliser les flux CDC dans Amazon Keyspaces
Pour activer les flux CDC, le principal, par exemple un utilisateur ou un rôle IAM, a besoin des autorisations suivantes.
Pour plus d'informations sur Gestion des identités et des accès AWS, voir[Gestion des identités et des accès AWS pour Amazon Keyspaces](security-iam.md).
## Autorisations permettant d'activer un flux CDC pour une table
[Pour activer un flux CDC pour une table Amazon Keyspaces, le principal a d'abord besoin des autorisations pour créer ou modifier une table, puis des autorisations pour créer le rôle AWSService RoleForAmazonKeyspaces CDC lié à un service.](using-service-linked-roles-CDC-streams.md#service-linked-role-permissions-CDC-streams) Amazon Keyspaces utilise le rôle lié au service pour publier des CloudWatch statistiques sur votre compte en votre nom
La politique IAM suivante en est un exemple.
```
{
"Version":"2012-10-17",
"Statement":[
{
"Effect":"Allow",
"Action":[
"cassandra:Create",
"cassandra:CreateMultiRegionResource",
"cassandra:Alter",
"cassandra:AlterMultiRegionResource"
],
"Resource":[
"arn:aws:cassandra:us-east-1:111122223333:/keyspace/my_keyspace/*",
"arn:aws:cassandra:us-east-1:111122223333:/keyspace/system*"
]
},
{
"Sid": "KeyspacesCDCServiceLinkedRole",
"Effect": "Allow",
"Action": "iam:CreateServiceLinkedRole",
"Resource": "arn:aws:iam::*:role/aws-service-role/cassandra-streams.amazonaws.com/AWSServiceRoleForAmazonKeyspacesCDC",
"Condition": {
"StringLike": {
"iam:AWSServiceName": "cassandra-streams.amazonaws.com"
}
}
}
]
}
```
Pour désactiver un flux, seules `ALTER TABLE` les autorisations sont requises.
## Autorisations pour visionner un stream du CDC
Pour afficher ou répertorier les flux CDC, le principal a besoin d'autorisations de lecture pour le keyspace du système. Pour de plus amples informations, veuillez consulter [`system_schema_mcs`](working-with-keyspaces.md#keyspace_system_schema_mcs).
La politique IAM suivante en est un exemple.
```
{
"Version":"2012-10-17",
"Statement":[
{
"Effect":"Allow",
"Action":"cassandra:Select",
"Resource":[
"arn:aws:cassandra:us-east-1:111122223333:/keyspace/system*"
]
}
]
}
```
Pour afficher ou répertorier les flux CDC à l'aide de l'API Amazon Keyspaces AWS CLI ou de l'API Amazon Keyspaces, le principal a besoin d'autorisations supplémentaires pour les actions `cassandra:ListStreams` et. `cassandra:GetStream`
La politique IAM suivante en est un exemple.
```
{
"Version": "2012-10-17",
"Statement": [
{
"Effect": "Allow",
"Action": [
"cassandra:Select",
"cassandra:ListStreams",
"cassandra:GetStream"
],
"Resource": "*"
}
]
}
```
## Autorisations pour lire un stream du CDC
Pour lire les flux du CDC, le directeur doit disposer des autorisations suivantes.
```
{
"Version":"2012-10-17",
"Statement":[
{
"Effect":"Allow",
"Action":[
"cassandra:GetStream",
"cassandra:GetShardIterator",
"cassandra:GetRecords"
],
"Resource":[
"arn:aws:cassandra:us-east-1:111122223333:/keyspace/my_keyspace/table/my_table/stream/stream_label"
]
}
]
}
```
## Autorisations pour traiter les flux CDC d'Amazon Keyspaces avec la Kinesis Client Library (KCL)
Pour traiter les flux CDC d'Amazon Keyspaces avec KCL, le principal IAM a besoin des autorisations suivantes.
+ `Amazon Keyspaces`— Accès en lecture seule à un flux Amazon Keyspaces CDC spécifique.
+ `DynamoDB`— Autorisations permettant de créer `shard lease` des tables, d'accéder en lecture et en écriture aux tables et d'accéder en lecture à l'index, conformément aux exigences du traitement des flux KCL.
+ `CloudWatch`— Autorisations permettant de publier des données métriques depuis Amazon Keyspaces CDC diffuse le traitement avec KCL dans l'espace de noms de votre application cliente KCL sur votre compte. CloudWatch Pour plus d'informations sur la surveillance, consultez [Surveiller la bibliothèque cliente Kinesis avec Amazon](https://docs.aws.amazon.com/streams/latest/dev/monitoring-with-kcl.html). CloudWatch
```
{
"Version":"2012-10-17",
"Statement":[
{
"Effect":"Allow",
"Action":[
"cassandra:GetStream",
"cassandra:GetShardIterator",
"cassandra:GetRecords"
],
"Resource":[
"arn:aws:cassandra:us-east-1:111122223333:/keyspace/my_keyspace/table/my_table/stream/stream_label"
]
},
{
"Effect":"Allow",
"Action":[
"dynamodb:CreateTable",
"dynamodb:DescribeTable",
"dynamodb:UpdateTable",
"dynamodb:GetItem",
"dynamodb:UpdateItem",
"dynamodb:PutItem",
"dynamodb:DeleteItem",
"dynamodb:Scan"
],
"Resource":[
"arn:aws:dynamodb:us-east-1:111122223333:table/KCL_APPLICATION_NAME"
]
},
{
"Effect":"Allow",
"Action":[
"dynamodb:CreateTable",
"dynamodb:DescribeTable",
"dynamodb:GetItem",
"dynamodb:UpdateItem",
"dynamodb:PutItem",
"dynamodb:DeleteItem",
"dynamodb:Scan"
],
"Resource":[
"arn:aws:dynamodb:us-east-1:111122223333:table/KCL_APPLICATION_NAME-WorkerMetricStats",
"arn:aws:dynamodb:us-east-1:111122223333:table/KCL_APPLICATION_NAME-CoordinatorState"
]
},
{
"Effect":"Allow",
"Action":[
"dynamodb:Query"
],
"Resource":[
"arn:aws:dynamodb:us-east-1:111122223333:table/KCL_APPLICATION_NAME/index/*"
]
},
{
"Effect":"Allow",
"Action":[
"cloudwatch:PutMetricData"
],
"Resource":"*"
}
]
}
```
# Comment accéder aux points de terminaison CDC Stream dans Amazon Keyspaces
Amazon Keyspaces gère des [points de terminaison](programmatic.endpoints.md#global_endpoints) distincts pour keyspaces/tables et pour les flux CDC dans chaque endroit où Région AWS Amazon Keyspaces est disponible. Pour accéder à un flux CDC, sélectionnez la région de la table et remplacez le `cassandra` préfixe par le nom du point de terminaison, comme indiqué dans l'exemple suivant : `cassandra-streams`
[\[See the AWS documentation website for more details\]](http://docs.aws.amazon.com/fr_fr/keyspaces/latest/devguide/CDC_access-endpoints.html)
Le tableau suivant contient une liste complète des points de terminaison publics disponibles pour Amazon Keyspaces change data capture streams. Amazon Keyspaces CDC streams prend en charge les deux IPv4 et IPv6. Tous les points de terminaison publics, par exemple`cassandra-streams.us-east-1.api.aws`, sont des points de terminaison à double pile qui peuvent être configurés pour IPv4 et. IPv6
[\[See the AWS documentation website for more details\]](http://docs.aws.amazon.com/fr_fr/keyspaces/latest/devguide/CDC_access-endpoints.html)
# Activer un flux CDC lors de la création d'une nouvelle table dans Amazon Keyspaces
Pour activer un flux CDC lorsque vous créez une table, vous pouvez utiliser l'`CREATE TABLE`instruction dans le CQL ou la `create-table` commande avec le AWS CLI.
Pour chaque ligne modifiée du tableau, Amazon Keyspaces peut enregistrer les modifications suivantes en fonction `view_type` de celle `cdc_specification` que vous avez sélectionnée :
+ `NEW_AND_OLD_IMAGES`— les deux versions de la ligne, avant et après la modification. Il s’agit de l’option par défaut.
+ `NEW_IMAGE`— la version de la ligne après la modification.
+ `OLD_IMAGE`— la version de la ligne avant la modification.
+ `KEYS_ONLY`— les clés de partition et de clustering de la ligne qui a été modifiée.
Pour plus d'informations sur la façon de baliser un flux, consultez[Ajouter des balises à un nouveau flux lors de la création d'une table](Tagging.Operations.new.table.stream.md).
**Note**
Amazon Keyspaces CDC nécessite la présence d'un rôle lié à un service (`AWSServiceRoleForAmazonKeyspacesCDC`) qui publie les données métriques provenant des flux Amazon Keyspaces CDC sur votre compte en votre nom`"cloudwatch:namespace": "AWS/Cassandra"`. CloudWatch Ce rôle est créé automatiquement pour vous. Pour de plus amples informations, veuillez consulter [Utilisation de rôles pour les flux CDC d'Amazon Keyspaces](using-service-linked-roles-CDC-streams.md).
------
#### [ Cassandra Query Language (CQL) ]
**Activez un flux CDC lorsque vous créez une table avec CQL**
1.
```
CREATE TABLE mykeyspace.mytable (a text, b text, PRIMARY KEY(a))
WITH CUSTOM_PROPERTIES={'cdc_specification': {'view_type': 'NEW_IMAGE'}} AND CDC = TRUE;
```
1. Pour confirmer les paramètres de diffusion, vous pouvez utiliser l'instruction suivante.
```
SELECT keyspace_name, table_name, cdc, custom_properties FROM system_schema_mcs.tables WHERE keyspace_name = 'mykeyspace' AND table_name = 'mytable';
```
Le résultat de cette instruction devrait ressembler à ceci.
```
SELECT keyspace_name, table_name, cdc, custom_properties FROM system_schema_mcs.tables WHERE keyspace_name = 'mykeyspace' AND table_name = 'mytable';keyspace_name | table_name | cdc | custom_properties
---------------+------------+------+-----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------
mykeyspace | mytable | True | {'capacity_mode': {'last_update_to_pay_per_request_timestamp': '1741383893782', 'throughput_mode': 'PAY_PER_REQUEST'}, 'cdc_specification': {'latest_stream_arn': 'arn:aws:cassandra:us-east-1:111122223333:/keyspace/mykeyspace/table/mytable/stream/2025-03-07T21:44:53.783', 'status': 'ENABLED', 'view_type': 'NEW_IMAGE'}, 'encryption_specification': {'encryption_type': 'AWS_OWNED_KMS_KEY'}, 'point_in_time_recovery': {'status': 'disabled'}}>
```
------
#### [ CLI ]
**Activez un flux CDC lorsque vous créez une table avec AWS CLI**
1. Pour créer un flux, vous pouvez utiliser la syntaxe suivante.
```
aws keyspaces create-table \
--keyspace-name 'mykeyspace' \
--table-name 'mytable' \
--schema-definition 'allColumns=[{name=a,type=text},{name=b,type=text}],partitionKeys=[{name=a}]' \
--cdc-specification status=ENABLED,viewType=NEW_IMAGE
```
1. La sortie de cette commande montre la `create-table` réponse standard et ressemble à cet exemple.
```
{ "resourceArn": "arn:aws:cassandra:us-east-1:111122223333:/keyspace/mykeyspace/table/mytable" }
```
------
# Activer un flux CDC pour une table existante dans Amazon Keyspaces
Pour activer un flux CDC pour une table existante, vous pouvez utiliser l'`ALTER TABLE`instruction dans CQL, la `update-table` commande avec le AWS CLI, ou vous pouvez utiliser la console.
Pour chaque ligne modifiée du tableau, Amazon Keyspaces peut enregistrer les modifications suivantes en fonction `view_type` de celle `cdc_specification` que vous avez sélectionnée :
+ `NEW_AND_OLD_IMAGES`— les deux versions de la ligne, avant et après la modification. Il s’agit de l’option par défaut.
+ `NEW_IMAGE`— la version de la ligne après la modification.
+ `OLD_IMAGE`— la version de la ligne avant la modification.
+ `KEYS_ONLY`— les clés de partition et de clustering de la ligne qui a été modifiée.
Pour plus d'informations sur la façon de baliser un flux, consultez[Ajouter de nouveaux tags à un stream](Tagging.Operations.existing.stream.md).
**Note**
Amazon Keyspaces CDC nécessite la présence d'un rôle lié à un service (`AWSServiceRoleForAmazonKeyspacesCDC`) qui publie les données métriques provenant des flux Amazon Keyspaces CDC sur votre compte en votre nom`"cloudwatch:namespace": "AWS/Cassandra"`. CloudWatch Ce rôle est créé automatiquement pour vous. Pour de plus amples informations, veuillez consulter [Utilisation de rôles pour les flux CDC d'Amazon Keyspaces](using-service-linked-roles-CDC-streams.md).
------
#### [ Cassandra Query Language (CQL) ]
**Activer un flux (flux CDC) avec CQL**
Vous pouvez l'utiliser `ALTER TABLE` pour activer un flux pour une table existante.
1. L'exemple suivant crée un flux qui capture uniquement les modifications apportées aux clés de partition et de clustering d'une ligne modifiée.
```
ALTER TABLE mykeyspace.mytable
WITH cdc = TRUE
AND CUSTOM_PROPERTIES={'cdc_specification': {'view_type': 'KEYS_ONLY'}};
```
1. Pour vérifier les paramètres de diffusion, vous pouvez utiliser l'instruction suivante.
```
SELECT keyspace_name, table_name, cdc, custom_properties FROM system_schema_mcs.tables WHERE keyspace_name = 'mykeyspace' AND table_name = 'mytable';
```
Le résultat de l'instruction ressemble à ceci.
```
keyspace_name | table_name | cdc | custom_properties
---------------+------------+------+--------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------
mykeyspace | mytable | True | {'capacity_mode': {'last_update_to_pay_per_request_timestamp': '1741385897045', 'throughput_mode': 'PAY_PER_REQUEST'}, 'cdc_specification': {'latest_stream_arn': 'arn:aws:cassandra:us-east-1:111122223333:/keyspace/mykeyspace/table/mytable/stream/2025-03-07T22:20:10.454', 'status': 'ENABLED', 'view_type': 'KEYS_ONLY'}, 'encryption_specification': {'encryption_type': 'AWS_OWNED_KMS_KEY'}, 'point_in_time_recovery': {'status': 'disabled'}}
```
------
#### [ CLI ]
**Créez un flux CDC avec le AWS CLI**
1. Pour créer un flux pour une table existante, vous pouvez utiliser la syntaxe suivante.
```
aws keyspaces update-table \
--keyspace-name 'mykeyspace' \
--table-name 'mytable' \
--cdc-specification status=ENABLED,viewType=NEW_AND_OLD_IMAGES
```
1. La sortie de cette commande montre la `create-table` réponse standard et ressemble à cet exemple.
```
{ "resourceArn": "arn:aws:cassandra:us-east-1:111122223333:/keyspace/mykeyspace/table/mytable" }
```
------
#### [ Console ]
**Activer un flux CDC avec la console Amazon Keyspaces**
1. [Connectez-vous à la AWS Management Console console Amazon Keyspaces et ouvrez-la chez https://console.aws.amazon.com/keyspaces/ vous.](https://console.aws.amazon.com/keyspaces/home)
1. Dans le volet de navigation, choisissez **Tables**, puis choisissez une table dans la liste.
1. Choisissez l'onglet **Streams**.
1. Choisissez **Modifier** pour activer un flux.
1. Sélectionnez **Activer les streams**.
1. Choisissez **le type d'affichage** du flux. Les options suivantes sont disponibles. Notez que vous ne pouvez pas modifier le type de vue d'un flux une fois celui-ci créé.
+ **Nouvelles et anciennes images** : Amazon Keyspaces capture les deux versions de la ligne, avant et après la modification. Il s’agit de l’option par défaut.
+ **Nouvelle image** — Amazon Keyspaces capture uniquement la version de la ligne après la modification.
+ **Ancienne image** — Amazon Keyspaces capture uniquement la version de la ligne avant la modification.
+ **Clé primaire uniquement** : Amazon Keyspaces capture uniquement les colonnes clés de partition et de clustering de la ligne modifiée.
1. Pour terminer, choisissez **Enregistrer les modifications**.
------
# Désactiver un flux CDC dans Amazon Keyspaces
Pour désactiver un flux CDC dans un espace de touches, vous pouvez utiliser l'`ALTER TABLE`instruction dans CQL, la `update-table` commande associée à AWS CLI, ou la console.
------
#### [ Cassandra Query Language (CQL) ]
**Désactiver un flux (flux CDC) avec CQL**
1. Pour désactiver un flux, vous pouvez utiliser l'instruction suivante.
```
ALTER TABLE mykeyspace.mytable
WITH cdc = FALSE;
```
1. Pour confirmer que le flux est désactivé, vous pouvez utiliser l'instruction suivante.
```
SELECT keyspace_name, table_name, cdc, custom_properties FROM system_schema_mcs.tables WHERE keyspace_name = 'mykeyspace' AND table_name = 'mytable';
```
Le résultat de cette instruction ressemble à ceci.
```
keyspace_name | table_name | cdc | custom_properties
---------------+------------+-------+------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------
mykeyspace | mytable | False | {'capacity_mode': {'last_update_to_pay_per_request_timestamp': '1741385668642', 'throughput_mode': 'PAY_PER_REQUEST'}, 'encryption_specification': {'encryption_type': 'AWS_OWNED_KMS_KEY'}, 'point_in_time_recovery': {'status': 'disabled'}}
```
------
#### [ CLI ]
**Désactiver un flux (flux CDC) avec le AWS CLI**
1. Pour désactiver un flux, vous pouvez utiliser la commande suivante.
```
aws keyspaces update-table \
--keyspace-name 'mykeyspace' \
--table-name 'mytable' \
--cdc-specification status=DISABLED
```
1. Le résultat de la commande ressemble à celui de cet exemple.
```
{
"keyspaceArn": "arn:aws:cassandra:us-east-1:111122223333:/keyspace/my_keyspace/",
"streamName": "my_stream"
}
```
------
#### [ Console ]
**Désactiver un flux (flux CDC) avec la console Amazon Keyspaces**
1. [Connectez-vous à la AWS Management Console console Amazon Keyspaces et ouvrez-la chez https://console.aws.amazon.com/keyspaces/ vous.](https://console.aws.amazon.com/keyspaces/home)
1. Dans le volet de navigation, choisissez **Tables**, puis choisissez une table dans la liste.
1. Choisissez l'onglet **Streams**.
1. Choisissez **Modifier**.
1. Désélectionnez **Activer les streams**.
1. Choisissez **Enregistrer les modifications** pour désactiver le flux.
------
# Afficher les streams du CDC sur Amazon Keyspaces
Pour afficher ou répertorier tous les flux dans l'espace clavier, vous pouvez interroger la table `system_schema_mcs.streams` dans l'espace clavier du système à l'aide d'une instruction dans CQL, ou utiliser les `list-stream` commandes `get-stream` et avec le ou la AWS CLI console.
Pour les autorisations requises, consultez [Configurer les autorisations pour utiliser les flux CDC dans Amazon Keyspaces](configure-cdc-permissions.md).
------
#### [ Cassandra Query Language (CQL) ]
**Afficher les flux du CDC avec CQL**
+ Pour contrôler l'état CDC de votre table, vous pouvez utiliser l'instruction suivante.
```
SELECT custom_properties
FROM system_schema_mcs.tables
WHERE keyspace_name='my_keyspace' and table_name='my_table';
```
Le résultat de la commande ressemble à ceci.
```
...
custom_properties
----------------------------------------------------------------------------------
{'cdc_specification':{'status': 'Enabled', 'view_type': 'NEW_IMAGE', 'latest_stream_arn': 'arn:aws:cassandra:us-east-1:111122223333:/keyspace/my_keyspace/table/my_table/stream/stream_label''}}
...
```
------
#### [ CLI ]
**Visionnez les diffusions du CDC avec le AWS CLI**
1. Cet exemple montre comment afficher les informations de flux d'une table.
```
aws keyspaces get-table \
--keyspace-name 'my_keyspace' \
--table-name 'my_table'
```
Le résultat de la commande ressemble à ceci.
```
{
"keyspaceName": "my_keyspace",
"tableName": "my_table",
... Other fields ...,
"latestStreamArn": "arn:aws:cassandra:us-east-1:111122223333:/keyspace/my_keyspace/table/my_table/stream/stream_label",
"cdcSpecification": {
"status": "ENABLED",
"viewType": "NEW_AND_OLD_IMAGES"
}
}
```
1. Vous pouvez répertorier tous les flux de votre compte dans un format spécifique Région AWS. La commande suivante en est un exemple.
```
aws keyspacesstreams list-streams --region us-east-1
```
Le résultat de la commande peut ressembler à ceci.
```
{
"Streams": [
{
"StreamArn": "arn:aws:cassandra:us-east-1:111122223333:/keyspace/ks_1/table/t1/stream/2023-05-11T21:21:33.291",
"StreamLabel": "2023-05-11T21:21:33.291",
"KeyspaceName": "ks_1"
"TableName": "t1",
},
{
"StreamArn": "arn:aws:cassandra:us-east-1:111122223333:/keyspace/ks_1/table/t2/stream/2023-05-11T21:21:33.291",
"StreamLabel": "2023-05-11T21:21:33.291",
"KeyspaceName": "ks_1"Create a keyspace with the name catalog. Note
that streams are not supported in multi-Region keyspaces.
"TableName": "t2",
},
{
"StreamArn": "arn:aws:cassandra:us-east-1:111122223333:/keyspace/ks_2/table/t1/stream/2023-05-11T21:21:33.291",
"StreamLabel": "2023-05-11T21:21:33.291",
"KeyspaceName": "ks_3"
"TableName": "t1",
}
]
}
```
1. Vous pouvez également répertorier les flux CDC pour un espace de touches donné à l'aide des paramètres suivants.
```
aws keyspacesstreams list-streams --keyspace-name ks_1 --region us-east-1
```
Le résultat de la commande ressemble à ceci.
```
{
"Streams": [
{
"StreamArn": "arn:aws:cassandra:us-east-1:111122223333:/keyspace/ks_1/table/t1/stream/2023-05-11T21:21:33.291",
"StreamLabel": "2023-05-11T21:21:33.291",
"KeyspaceName": "ks_1"
"TableName": "t1",
},
{
"StreamArn": "arn:aws:cassandra:us-east-1:111122223333:/keyspace/ks_1/table/t2/stream/2023-05-11T21:21:33.291",
"StreamLabel": "2023-05-11T21:21:33.291",
"KeyspaceName": "ks_1"
"TableName": "t2",
}
]
}
```
1. Vous pouvez également répertorier les flux CDC pour une table donnée à l'aide des paramètres suivants.
```
aws keyspacesstreams list-streams --keyspace-name ks_1 --table-name t2 --region us-east-1
```
Le résultat de la commande ressemble à ceci.
```
{
"Streams": [
{
"StreamArn": "arn:aws:cassandra:us-east-1:111122223333:/keyspace/ks_1/table/t2/stream/2023-05-11T21:21:33.291",
"StreamLabel": "2023-05-11T21:21:33.291",
"KeyspaceName": "ks_1"
"TableName": "t2",
}
]
}
```
------
#### [ Console ]
**Afficher les flux CDC dans la console Amazon Keyspaces**
1. [Connectez-vous à la AWS Management Console console Amazon Keyspaces et ouvrez-la chez https://console.aws.amazon.com/keyspaces/ vous.](https://console.aws.amazon.com/keyspaces/home)
1. Dans le volet de navigation, choisissez **Tables**, puis choisissez une table dans la liste.
1. Choisissez l'onglet **Streams** pour consulter les détails du stream.
------
# Accédez aux enregistrements des flux CDC sur Amazon Keyspaces
Pour accéder aux enregistrements d'un flux, vous utilisez l'API [Amazon Keyspaces Streams](https://docs.aws.amazon.com/keyspaces/latest/StreamsAPIReference/Welcome.html). La section suivante contient des exemples sur la manière d'accéder aux enregistrements à l'aide du AWS CLI.
Pour les autorisations requises, consultez [Configurer les autorisations pour utiliser les flux CDC dans Amazon Keyspaces](configure-cdc-permissions.md).
**Accédez aux enregistrements d'un flux à l'aide du AWS CLI**
1. Vous pouvez utiliser l'API Amazon Keyspaces Streams pour accéder aux enregistrements de modifications du flux. Pour plus d'informations, consultez le manuel de référence de l'[https://docs.aws.amazon.com/keyspaces/latest/StreamsAPIReference/Welcome.html](https://docs.aws.amazon.com/keyspaces/latest/StreamsAPIReference/Welcome.html). Pour récupérer les fragments contenus dans le flux, vous pouvez utiliser l'`get-stream`API comme indiqué dans l'exemple suivant.
```
aws keyspacesstreams get-stream \
--stream-arn 'arn:aws:cassandra:us-east-1:111122223333:/keyspace/mykeyspace/table/mytable/stream/STREAM_LABEL'
```
Voici un exemple de la sortie.
```
{
"StreamArn": "arn:aws:cassandra:us-east-1:111122223333:/keyspace/mykeyspace/table/mytable/stream/2023-05-11T21:21:33.291",
"StreamStatus": "ENABLED",
"StreamViewType": "NEW_AND_OLD_IMAGES",
"CreationRequestDateTime": "",
"KeyspaceName": "mykeyspace",
"TableName": "mytable",
"StreamLabel": "2023-05-11T21:21:33.291",
"Shards": [
{
"SequenceNumberRange": {
"EndingSequenceNumber": "",
"StartingSequenceNumber": ""
},
"ShardId": ""
},
]
}
```
1. Pour récupérer des enregistrements du flux, vous devez commencer par obtenir un itérateur qui vous fournit le point de départ pour accéder aux enregistrements. Pour ce faire, vous pouvez utiliser les fragments contenus dans le flux CDC renvoyé par l'API à l'étape précédente. Pour rassembler l'itérateur, vous pouvez utiliser l'`get-shard-iterator`API. Dans cet exemple, vous utilisez un itérateur de type `TRIM_HORIZON` qui extrait à partir du dernier point découpé (ou début) de la partition.
```
aws keyspacesstreams get-shard-iterator \
--stream-arn 'arn:aws:cassandra:us-east-1:111122223333:/keyspace/mykeyspace/table/mytable/stream/STREAM_LABEL' \
--shard-id 'SHARD_ID' \
--shard-iterator-type 'TRIM_HORIZON'
```
Le résultat de la commande ressemble à celui de l'exemple suivant.
```
{
"ShardIterator": ""
}
```
1. Pour récupérer les enregistrements CDC à l'aide de l'`get-records`API, vous pouvez utiliser l'itérateur renvoyé à la dernière étape. La commande suivante en est un exemple.
```
aws keyspacesstreams get-records \
--shard-iterator 'SHARD_ITERATOR' \
--limit 100
```
# Utiliser la bibliothèque client Kinesis (KCL) pour traiter les flux Amazon Keyspaces
Cette rubrique explique comment utiliser la bibliothèque client Kinesis (KCL) pour consommer et traiter les données issues des flux de capture des données de modification (CDC) d'Amazon Keyspaces.
Au lieu de travailler directement avec l'API Amazon Keyspaces Streams, l'utilisation de la bibliothèque client Kinesis (KCL) offre de nombreux avantages, par exemple :
+ Suivi du lignage des fragments et gestion des itérateurs intégrés.
+ Équilibrage automatique de la charge entre les travailleurs.
+ Tolérance aux pannes et rétablissement en cas de défaillance d'un travailleur.
+ Pointage de points de contrôle pour suivre la progression du traitement.
+ Adaptation aux changements de capacité du cours d'eau.
+ Informatique distribuée simplifiée pour le traitement des enregistrements du CDC.
La section suivante explique pourquoi et comment utiliser la bibliothèque client Kinesis (KCL) pour traiter des flux et fournit un exemple de traitement d'un flux CDC Amazon Keyspaces avec la KCL.
Pour plus d'informations sur les tarifs, consultez les tarifs d'[Amazon Keyspaces (pour Apache Cassandra](https://aws.amazon.com/keyspaces/pricing)).
## Qu'est-ce que Kinesis Client Library ?
La Kinesis Client Library (KCL) est une bibliothèque logicielle Java autonome conçue pour simplifier le processus de consommation et de traitement des données issues des flux. KCL gère de nombreuses tâches complexes associées à l'informatique distribuée, ce qui vous permet de vous concentrer sur la mise en œuvre de votre logique métier lors du traitement des données de flux. KCL gère des activités telles que l'équilibrage de charge entre plusieurs travailleurs, la réponse aux défaillances des travailleurs, le contrôle des enregistrements traités et la réponse aux modifications du nombre de partitions dans le flux.
Pour traiter les flux CDC d'Amazon Keyspaces, vous pouvez utiliser les modèles de conception trouvés dans la KCL pour travailler avec des fragments de flux et des enregistrements de flux. La KCL simplifie le codage en fournissant des abstractions utiles par-dessus l’API Kinesis Data Streams de bas niveau. Pour plus d'informations sur la KCL, consultez la section [Develop consumers with KCL](https://docs.aws.amazon.com/kinesis/latest/dev/develop-kcl-consumers.html) dans le manuel *Amazon Kinesis Data* Streams Developer Guide.
Pour écrire des applications à l'aide de la KCL, vous utilisez l'adaptateur Amazon Keyspaces Streams Kinesis. L'adaptateur Kinesis implémente l'interface Kinesis Data Streams afin que vous puissiez utiliser la KCL pour consommer et traiter les enregistrements des flux Amazon Keyspaces. Pour obtenir des instructions sur la configuration et l'installation de l'adaptateur Amazon Keyspaces Streams Kinesis, consultez le référentiel. [GitHub](https://github.com/aws/keyspaces-streams-kinesis-adapter)
Le schéma suivant montre comment ces bibliothèques interagissent les unes avec les autres.
![\[Interaction entre les applications d'un client et Kinesis Data Streams, KCL, l'adaptateur Amazon Keyspaces Streams Kinesis et Amazon Keyspaces APIs lors du traitement des enregistrements de flux Amazon Keyspaces CDC.\]](http://docs.aws.amazon.com/fr_fr/keyspaces/latest/devguide/images/keyspaces-streams-kinesis-adapter.png)
KCL est fréquemment mis à jour pour intégrer de nouvelles versions des bibliothèques sous-jacentes, des améliorations de sécurité et des corrections de bogues. Nous vous recommandons d'utiliser la dernière version de KCL pour éviter les problèmes connus et bénéficier de toutes les dernières améliorations. Pour trouver la dernière version de KCL, consultez le référentiel [KCL GitHub ](https://github.com/awslabs/amazon-kinesis-client).
## Concepts du KCL
Avant d'implémenter une application grand public à l'aide de KCL, vous devez comprendre les concepts suivants :
**Application KCL destinée aux consommateurs**
Une application client KCL est un programme qui traite les données d'un flux Amazon Keyspaces CDC. Le KCL sert d'intermédiaire entre le code de votre application client et le flux Amazon Keyspaces CDC.
**Travailleur**
Un worker est une unité d'exécution de votre application client KCL qui traite les données du flux CDC Amazon Keyspaces. Votre application peut exécuter plusieurs travailleurs répartis sur plusieurs instances.
**Processeur d'enregistrement**
Un processeur d'enregistrements est la logique de votre application qui traite les données d'une partition dans le flux CDC d'Amazon Keyspaces. Un processeur d'enregistrements est instancié par un travailleur pour chaque partition qu'il gère.
**Bail**
Un bail représente la responsabilité du traitement d'une partition. Les travailleurs utilisent les baux pour coordonner quel travailleur traite quel fragment. KCL stocke les données de location dans une table dans Amazon DynamoDB.
**Point de contrôle**
Un point de contrôle est un enregistrement de la position dans la partition jusqu'à laquelle le processeur d'enregistrements a traité avec succès les enregistrements. Le point de contrôle permet à votre demande de reprendre le traitement là où il s'était arrêté en cas d'échec d'un collaborateur.
Une fois l'adaptateur Amazon Keyspaces Kinesis en place, vous pouvez commencer à développer depuis l'interface KCL, les appels d'API étant directement dirigés vers le point de terminaison du stream Amazon Keyspaces. Pour obtenir la liste des points de terminaison disponibles, reportez-vous à la section [Comment accéder aux points de terminaison CDC Stream dans Amazon Keyspaces](CDC_access-endpoints.md).
Lorsque votre application démarre, elle appelle la KCL pour instancier un worker. Vous devez fournir au travailleur les informations de configuration de l'application, telles que le descripteur de flux et les AWS informations d'identification, ainsi que le nom d'une classe de processeur d'enregistrements que vous fournissez. À mesure qu’il exécute le code dans le processeur d’enregistrements, le worker effectue les tâches suivantes :
+ Se connecte au flux
+ Énumère les partitions dans le flux.
+ Coordonne les associations de partition avec les autres travaux (le cas échéant)
+ Instancie un processeur d’enregistrements pour chaque partition qu’il gère
+ Extrait des enregistrements du flux
+ Pousse les enregistrements sur le processeur d’enregistrements correspondant
+ Contrôle les enregistrements traités
+ Équilibre les associations partition-worker lorsque le nombre d’instances de worker change
+ Équilibre les associations partition-worker quand des partitions sont fractionnées
# Implémenter une application client KCL pour les flux CDC d'Amazon Keyspaces
Cette rubrique fournit un step-by-step guide pour implémenter une application client KCL pour traiter les flux CDC d'Amazon Keyspaces.
1. Conditions préalables : Avant de commencer, assurez-vous d'avoir :
+ Une table Amazon Keyspaces avec un flux CDC
+ Autorisations IAM requises pour que le principal IAM puisse accéder au flux CDC d'Amazon Keyspaces, créer et accéder aux tables DynamoDB pour le traitement des flux KCL, et autorisations pour publier des métriques sur. CloudWatch Pour plus d'informations et un exemple de stratégie, consultez[Autorisations pour traiter les flux CDC d'Amazon Keyspaces avec la Kinesis Client Library (KCL)](configure-cdc-permissions.md#cdc-permissions-kcl).
+ Assurez-vous que des AWS informations d'identification valides sont configurées dans votre configuration locale. Pour de plus amples informations, veuillez consulter [Stocker les clés d'accès pour un accès programmatique](aws.credentials.manage.md).
+ Kit de développement Java (JDK) 8 ou version ultérieure
+ Les exigences sont répertoriées dans le [fichier Readme](https://github.com/aws/keyspaces-streams-kinesis-adapter) sur Github.
1. Au cours de cette étape, vous ajoutez la dépendance KCL à votre projet. Pour Maven, ajoutez ce qui suit à votre fichier pom.xml :
```
software.amazon.kinesis
amazon-kinesis-client
3.1.0
software.amazon.keyspaces
keyspaces-streams-kinesis-adapter
1.0.0
```
**Note**
Vérifiez toujours la dernière version de KCL dans le référentiel [KCL GitHub ](https://github.com/awslabs/amazon-kinesis-client).
1. Créez une classe d'usine qui produit des instances de processeur d'enregistrement :
```
import software.amazon.awssdk.services.keyspacesstreams.model.Record;
import software.amazon.keyspaces.streamsadapter.adapter.KeyspacesStreamsClientRecord;
import software.amazon.keyspaces.streamsadapter.model.KeyspacesStreamsProcessRecordsInput;
import software.amazon.keyspaces.streamsadapter.processor.KeyspacesStreamsShardRecordProcessor;
import software.amazon.kinesis.lifecycle.events.InitializationInput;
import software.amazon.kinesis.lifecycle.events.LeaseLostInput;
import software.amazon.kinesis.lifecycle.events.ShardEndedInput;
import software.amazon.kinesis.lifecycle.events.ShutdownRequestedInput;
import software.amazon.kinesis.processor.RecordProcessorCheckpointer;
public class RecordProcessor implements KeyspacesStreamsShardRecordProcessor {
private String shardId;
@Override
public void initialize(InitializationInput initializationInput) {
this.shardId = initializationInput.shardId();
System.out.println("Initializing record processor for shard: " + shardId);
}
@Override
public void processRecords(KeyspacesStreamsProcessRecordsInput processRecordsInput) {
try {
for (KeyspacesStreamsClientRecord record : processRecordsInput.records()) {
Record keyspacesRecord = record.getRecord();
System.out.println("Received record: " + keyspacesRecord);
}
if (!processRecordsInput.records().isEmpty()) {
RecordProcessorCheckpointer checkpointer = processRecordsInput.checkpointer();
try {
checkpointer.checkpoint();
System.out.println("Checkpoint successful for shard: " + shardId);
} catch (Exception e) {
System.out.println("Error while checkpointing for shard: " + shardId + " " + e);
}
}
} catch (Exception e) {
System.out.println("Error processing records for shard: " + shardId + " " + e);
}
}
@Override
public void leaseLost(LeaseLostInput leaseLostInput) {
System.out.println("Lease lost for shard: " + shardId);
}
@Override
public void shardEnded(ShardEndedInput shardEndedInput) {
System.out.println("Shard ended: " + shardId);
try {
// This is required. Checkpoint at the end of the shard
shardEndedInput.checkpointer().checkpoint();
System.out.println("Final checkpoint successful for shard: " + shardId);
} catch (Exception e) {
System.out.println("Error while final checkpointing for shard: " + shardId + " " + e);
throw new RuntimeException("Error while final checkpointing", e);
}
}
@Override
public void shutdownRequested(ShutdownRequestedInput shutdownRequestedInput) {
System.out.println("Shutdown requested for shard " + shardId);
try {
shutdownRequestedInput.checkpointer().checkpoint();
} catch (Exception e) {
System.out.println("Error while checkpointing on shutdown for shard: " + shardId + " " + e);
}
}
}
```
1. Créez une fabrique de disques comme indiqué dans l'exemple suivant.
```
import software.amazon.kinesis.processor.ShardRecordProcessor;
import software.amazon.kinesis.processor.ShardRecordProcessorFactory;
import java.util.Queue;
import java.util.concurrent.ConcurrentLinkedQueue;
public class RecordProcessorFactory implements ShardRecordProcessorFactory {
private final Queue processors = new ConcurrentLinkedQueue<>();
@Override
public ShardRecordProcessor shardRecordProcessor() {
System.out.println("Creating new RecordProcessor");
RecordProcessor processor = new RecordProcessor();
processors.add(processor);
return processor;
}
}
```
1. Au cours de cette étape, vous créez la classe de base à configurer KCLv3 et l'adaptateur Amazon Keyspaces.
```
import com.example.KCLExample.utils.RecordProcessorFactory;
import software.amazon.keyspaces.streamsadapter.AmazonKeyspacesStreamsAdapterClient;
import software.amazon.keyspaces.streamsadapter.StreamsSchedulerFactory;
import java.util.Arrays;
import java.util.List;
import java.util.concurrent.ExecutionException;
import software.amazon.awssdk.regions.Region;
import software.amazon.awssdk.services.cloudwatch.CloudWatchAsyncClient;
import software.amazon.awssdk.services.dynamodb.DynamoDbAsyncClient;
import software.amazon.awssdk.services.dynamodb.model.DeleteTableRequest;
import software.amazon.awssdk.services.dynamodb.model.DeleteTableResponse;
import software.amazon.awssdk.services.keyspacesstreams.KeyspacesStreamsClient;
import software.amazon.awssdk.services.kinesis.KinesisAsyncClient;
import software.amazon.kinesis.common.ConfigsBuilder;
import software.amazon.kinesis.common.InitialPositionInStream;
import software.amazon.kinesis.common.InitialPositionInStreamExtended;
import software.amazon.kinesis.coordinator.CoordinatorConfig;
import software.amazon.kinesis.coordinator.Scheduler;
import software.amazon.kinesis.leases.LeaseManagementConfig;
import software.amazon.kinesis.processor.ProcessorConfig;
import software.amazon.kinesis.processor.StreamTracker;
import software.amazon.kinesis.retrieval.polling.PollingConfig;
public class KCLTestBase {
protected KeyspacesStreamsClient streamsClient;
protected KinesisAsyncClient adapterClient;
protected DynamoDbAsyncClient dynamoDbAsyncClient;
protected CloudWatchAsyncClient cloudWatchClient;
protected Region region;
protected RecordProcessorFactory recordProcessorFactory;
protected Scheduler scheduler;
protected Thread schedulerThread;
public void baseSetUp() {
recordProcessorFactory = new RecordProcessorFactory();
setupKCLBase();
}
protected void setupKCLBase() {
region = Region.US_EAST_1;
streamsClient = KeyspacesStreamsClient.builder()
.region(region)
.build();
adapterClient = new AmazonKeyspacesStreamsAdapterClient(
streamsClient,
region);
dynamoDbAsyncClient = DynamoDbAsyncClient.builder()
.region(region)
.build();
cloudWatchClient = CloudWatchAsyncClient.builder()
.region(region)
.build();
}
protected void startScheduler(Scheduler scheduler) {
this.scheduler = scheduler;
schedulerThread = new Thread(() -> scheduler.run());
schedulerThread.start();
}
protected void shutdownScheduler() {
if (scheduler != null) {
scheduler.shutdown();
try {
schedulerThread.join(30000);
} catch (InterruptedException e) {
System.out.println("Error while shutting down scheduler " + e);
}
}
}
protected Scheduler createScheduler(String streamArn, String leaseTableName) {
String workerId = "worker-" + System.currentTimeMillis();
// Create ConfigsBuilder
ConfigsBuilder configsBuilder = createConfigsBuilder(streamArn, workerId, leaseTableName);
// Configure retrieval config for polling
PollingConfig pollingConfig = new PollingConfig(streamArn, adapterClient);
// Create the Scheduler
return StreamsSchedulerFactory.createScheduler(
configsBuilder.checkpointConfig(),
configsBuilder.coordinatorConfig(),
configsBuilder.leaseManagementConfig(),
configsBuilder.lifecycleConfig(),
configsBuilder.metricsConfig(),
configsBuilder.processorConfig(),
configsBuilder.retrievalConfig().retrievalSpecificConfig(pollingConfig),
streamsClient,
region
);
}
private ConfigsBuilder createConfigsBuilder(String streamArn, String workerId, String leaseTableName) {
ConfigsBuilder configsBuilder = new ConfigsBuilder(
streamArn,
leaseTableName,
adapterClient,
dynamoDbAsyncClient,
cloudWatchClient,
workerId,
recordProcessorFactory);
configureCoordinator(configsBuilder.coordinatorConfig());
configureLeaseManagement(configsBuilder.leaseManagementConfig());
configureProcessor(configsBuilder.processorConfig());
configureStreamTracker(configsBuilder, streamArn);
return configsBuilder;
}
private void configureCoordinator(CoordinatorConfig config) {
config.skipShardSyncAtWorkerInitializationIfLeasesExist(true)
.parentShardPollIntervalMillis(1000)
.shardConsumerDispatchPollIntervalMillis(500);
}
private void configureLeaseManagement(LeaseManagementConfig config) {
config.shardSyncIntervalMillis(0)
.leasesRecoveryAuditorInconsistencyConfidenceThreshold(0)
.leasesRecoveryAuditorExecutionFrequencyMillis(5000)
.leaseAssignmentIntervalMillis(1000L);
}
private void configureProcessor(ProcessorConfig config) {
config.callProcessRecordsEvenForEmptyRecordList(true);
}
private void configureStreamTracker(ConfigsBuilder configsBuilder, String streamArn) {
StreamTracker streamTracker = StreamsSchedulerFactory.createSingleStreamTracker(
streamArn,
InitialPositionInStreamExtended.newInitialPosition(InitialPositionInStream.TRIM_HORIZON)
);
configsBuilder.streamTracker(streamTracker);
}
public void deleteAllDdbTables(String baseTableName) {
List tablesToDelete = Arrays.asList(
baseTableName,
baseTableName + "-CoordinatorState",
baseTableName + "-WorkerMetricStats"
);
for (String tableName : tablesToDelete) {
deleteTable(tableName);
}
}
private void deleteTable(String tableName) {
DeleteTableRequest deleteTableRequest = DeleteTableRequest.builder()
.tableName(tableName)
.build();
try {
DeleteTableResponse response = dynamoDbAsyncClient.deleteTable(deleteTableRequest).get();
System.out.println("Table deletion response " + response);
} catch (InterruptedException | ExecutionException e) {
System.out.println("Error deleting table: " + tableName + " " + e);
}
}
}
```
1. Au cours de cette étape, vous implémentez la classe de processeur d'enregistrements pour que votre application commence à traiter les événements de modification.
```
import software.amazon.kinesis.coordinator.Scheduler;
public class KCLTest {
private static final int APP_RUNTIME_SECONDS = 1800;
private static final int SLEEP_INTERNAL_MS = 60*1000;
public static void main(String[] args) {
KCLTestBase kclTestBase;
kclTestBase = new KCLTestBase();
kclTestBase.baseSetUp();
// Create and start scheduler
String leaseTableName = generateUniqueApplicationName();
// Update below to your Stream ARN
String streamArn = "arn:aws:cassandra:us-east-1:759151643516:/keyspace/cdc_sample_test/table/test_kcl_bool/stream/2025-07-01T15:52:57.529";
Scheduler scheduler = kclTestBase.createScheduler(streamArn, leaseTableName);
kclTestBase.startScheduler(scheduler);
// Wait for specified time before shutting down - KCL applications are designed to run forever, however in this
// example we will shut it down after APP_RUNTIME_SECONDS
long startTime = System.currentTimeMillis();
long endTime = startTime + (APP_RUNTIME_SECONDS * 1000);
while (System.currentTimeMillis() < endTime) {
try {
// Print and sleep every minute
Thread.sleep(SLEEP_INTERNAL_MS);
System.out.println("Application is running");
} catch (InterruptedException e) {
System.out.println("Interrupted while waiting for records");
Thread.currentThread().interrupt();
break;
}
}
// Stop the scheduler
kclTestBase.shutdownScheduler();
kclTestBase.deleteAllDdbTables(leaseTableName);
}
public static String generateUniqueApplicationName() {
String timestamp = String.valueOf(System.currentTimeMillis());
String randomString = java.util.UUID.randomUUID().toString().substring(0, 8);
return String.format("KCL-App-%s-%s", timestamp, randomString);
}
}
```
## Bonnes pratiques
Suivez ces bonnes pratiques lorsque vous utilisez KCL avec les flux CDC d'Amazon Keyspaces :
**Gestion des erreurs**
Implémentez une gestion robuste des erreurs dans votre processeur d'enregistrements afin de gérer les exceptions avec élégance. Envisagez d'implémenter une logique de nouvelle tentative pour les échecs transitoires.
**Fréquence de pointage**
Équilibrez la fréquence des points de contrôle afin de minimiser le double traitement tout en garantissant un suivi raisonnable des progrès. Des points de contrôle trop fréquents peuvent avoir un impact sur les performances, tandis que des points de contrôle trop rares peuvent entraîner un retraitement supplémentaire en cas de défaillance d'un travailleur.
**Dimensionnement des effectifs**
Dimensionnez le nombre de travailleurs en fonction du nombre de fragments dans votre flux CDC. Un bon point de départ est d'avoir un travailleur par partition, mais vous devrez peut-être procéder à des ajustements en fonction de vos exigences de traitement.
**Surveillance**
Utilisez CloudWatch les indicateurs fournis par KCL pour surveiller l'état et les performances de votre application grand public. Les indicateurs clés incluent la latence de traitement, l'âge des points de contrôle et le nombre de baux.
**Test**
Testez votre application grand public de manière approfondie, notamment en cas de défaillance d'un opérateur, de repartage de flux et de conditions de charge variables.
## Utilisation de KCL avec des langages autres que Java
Bien que KCL soit principalement une bibliothèque Java, vous pouvez l'utiliser avec d'autres langages de programmation via le MultiLangDaemon. MultiLangDaemon Il s'agit d'un démon basé sur Java qui gère l'interaction entre votre processeur d'enregistrement non Java et le KCL.
KCL fournit un support pour les langues suivantes :
+ Python
+ Ruby
+ Node.js
+ .NET
Pour plus d'informations sur l'utilisation de KCL avec des langages autres que Java, consultez la documentation [ MultiLangDaemon KCL](https://github.com/awslabs/amazon-kinesis-client/tree/master/amazon-kinesis-client-multilang).
## Résolution des problèmes
Cette section fournit des solutions aux problèmes courants que vous pouvez rencontrer lors de l'utilisation de KCL avec les flux CDC d'Amazon Keyspaces.
**Traitement lent**
Si votre application client traite les dossiers lentement, pensez à :
+ Augmenter le nombre d'instances de travailleurs
+ Optimisation de votre logique de traitement des dossiers
+ Vérification de l'absence de goulots d'étranglement dans les systèmes en aval
**Traitement des doublons**
Si vous constatez un traitement dupliqué des enregistrements, vérifiez votre logique de pointage des points de contrôle. Assurez-vous de vérifier après avoir traité les enregistrements avec succès.
**Défaillances des travailleurs**
Si les travailleurs échouent fréquemment, vérifiez :
+ Contraintes de ressources (processeur, mémoire)
+ Problèmes liés à la connectivité réseau
+ Problèmes d’autorisations
**Problèmes liés aux tables de location**
Si vous rencontrez des problèmes avec le tableau des baux KCL :
+ Vérifiez que votre application dispose des autorisations appropriées pour accéder au tableau Amazon Keyspaces
+ Vérifiez que le débit provisionné de la table est suffisant