Die vorliegende Übersetzung wurde maschinell erstellt. Im Falle eines Konflikts oder eines Widerspruchs zwischen dieser übersetzten Fassung und der englischen Fassung (einschließlich infolge von Verzögerungen bei der Übersetzung) ist die englische Fassung maßgeblich.
# Arbeiten mit Change Data Capture (CDC) -Streams in Amazon Keyspaces
Amazon Keyspaces Change Data Capture (CDC) zeichnet Änderungsereignisse auf Zeilenebene aus einer Amazon Keyspaces-Tabelle nahezu in Echtzeit auf.
Amazon Keyspaces CDC ermöglicht ereignisgesteuerte Anwendungsfälle wie industrielles IoT und Betrugserkennung sowie Anwendungsfälle für die Datenverarbeitung wie Volltextsuche und Datenarchivierung. Die Änderungsereignisse, die Amazon Keyspaces CDC in Streams erfasst, können von nachgelagerten Anwendungen verarbeitet werden, die geschäftskritische Funktionen wie Datenanalyse, Textsuche, ML-Training/-Inferenz und kontinuierliche Datensicherungen für die Archivierung ausführen. Sie können beispielsweise Stream-Daten zur weiteren Verarbeitung an AWS Analyse- und Speicherdienste wie Amazon OpenSearch Service, Amazon Redshift und Amazon S3 übertragen.
Amazon Keyspaces CDC bietet zeitlich geordnete und deduplizierte Änderungsdatensätze für Tabellen mit automatischer Skalierung des Datendurchsatzes und einer Aufbewahrungszeit von bis zu 24 Stunden.
Amazon Keyspaces CDC-Streams sind vollständig serverlos, und Sie müssen die Dateninfrastruktur für die Erfassung von Änderungsereignissen nicht verwalten. Darüber hinaus verbraucht Amazon Keyspaces CDC keine Tabellenkapazität, weder für die Datenverarbeitung noch für die Speicherung. Weitere Informationen finden Sie unter [So funktionieren Change Data Capture (CDC) -Streams in Amazon Keyspaces](cdc_how-it-works.md).
Sie können die Amazon Keyspaces Streams-API verwenden, um Anwendungen zu erstellen, die Amazon Keyspaces CDC-Streams verwenden, und auf der Grundlage der Inhalte Maßnahmen ergreifen. Informationen zu verfügbaren Endpunkten finden Sie unter. [So greifen Sie in Amazon Keyspaces auf CDC-Stream-Endpunkte zu](CDC_access-endpoints.md)
Eine vollständige Liste aller Operationen, die für Amazon Keyspaces in der Streams-API verfügbar sind, finden Sie unter [https://docs.aws.amazon.com/keyspaces/latest/StreamsAPIReference/Welcome.html](https://docs.aws.amazon.com/keyspaces/latest/StreamsAPIReference/Welcome.html).
**Topics**
+ [So funktionieren Change Data Capture (CDC) -Streams in Amazon Keyspaces](cdc_how-it-works.md)
+ [So verwenden Sie Change Data Capture (CDC) -Streams in Amazon Keyspaces](cdc_how-to-use.md)
# So funktionieren Change Data Capture (CDC) -Streams in Amazon Keyspaces
Dieser Abschnitt bietet einen Überblick darüber, wie Change Data Capture (CDC) -Streams in Amazon Keyspaces funktionieren.
Amazon Keyspaces Change Data Capture (CDC) zeichnet eine geordnete Abfolge von Änderungen auf Zeilenebene in Amazon Keyspaces-Tabellen auf und speichert diese Informationen bis zu 24 Stunden in einem Protokoll namens *Stream*. Jede Änderung auf Zeilenebene generiert einen neuen CDC-Datensatz, der die Informationen zur Primärschlüsselspalte sowie den Status „Vorher“ und „Nachher“ der Zeile einschließlich aller Spalten enthält. Anwendungen können auf den Stream zugreifen und die Mutationen nahezu in Echtzeit anzeigen.
Wenn Sie CDC für Ihre Tabelle aktivieren, erstellt Amazon Keyspaces einen neuen CDC-Stream und beginnt, Informationen über jede Änderung in der Tabelle zu erfassen. Der CDC-Stream hat einen Amazon-Ressourcennamen (ARN) mit dem folgenden Format:
```
arn:${Partition}:cassandra:{Region}:${Account}:/keyspace/${keyspaceName}/table/${tableName}/stream/${streamLabel}
```
Sie können den Informationstyp oder den *Ansichtstyp* auswählen, den der CDC-Stream für jeden Datensatz sammelt, wenn Sie den CDC-Stream zum ersten Mal aktivieren. Sie können den Ansichtstyp des Streams anschließend nicht mehr ändern. Amazon Keyspaces unterstützt die folgenden Ansichtstypen:
+ `NEW_AND_OLD_IMAGES`— Erfasst die Versionen der Zeile vor und nach der Mutation. Das ist die Standardeinstellung.
+ `NEW_IMAGE`— Erfasst die Version der Zeile nach der Mutation.
+ `OLD_IMAGE`— Erfasst die Version der Zeile vor der Mutation.
+ `KEYS_ONLY`— Erfasst die Partitions- und Clusterschlüssel der Zeile, die mutiert wurde.
Jeder CDC-Stream besteht aus Datensätzen. Jeder Datensatz steht für eine einzelne Zeilenänderung in einer Amazon Keyspaces-Tabelle. *Datensätze sind logisch in Gruppen organisiert, die als Shards bezeichnet werden.* Diese Gruppen sind logisch nach Bereichen des Primärschlüssels (Kombination aus Partitionsschlüssel, Clusterschlüsselbereichen) organisiert und sind ein internes Konstrukt von Amazon Keyspaces. Jeder Shard fungiert als Container für mehrere Datensätze und enthält Informationen, die für den Zugriff auf diese Datensätze und die Iteration durch diese Datensätze erforderlich sind.
![\[Ein Amazon Keyspaces CDC-Stream besteht aus Shards, die einen CDC-Datensatz einer Sammlung von Zeilenmutationen darstellen.\]](http://docs.aws.amazon.com/de_de/keyspaces/latest/devguide/images/keyspaces_cdc.png)
Jedem CDC-Datensatz wird eine Sequenznummer zugewiesen, die die Reihenfolge widerspiegelt, in der der Datensatz innerhalb des Shards veröffentlicht wurde. Es ist garantiert, dass die Sequenznummer zunimmt und innerhalb jedes Shards einzigartig ist.
Amazon Keyspaces erstellt und löscht Shards automatisch. Basierend auf der Traffic-Auslastung kann Amazon Keyspaces auch Shards im Laufe der Zeit aufteilen oder zusammenführen. Amazon Keyspaces kann beispielsweise einen Shard in mehrere neue Shards aufteilen oder Shards zu einem neuen einzelnen Shard zusammenführen. Amazon Keyspaces APIs veröffentlicht die Shard- und CDC-Stream-Informationen, damit verbrauchende Anwendungen Datensätze in der richtigen Reihenfolge verarbeiten können, indem sie auf das gesamte Lineage-Diagramm eines Shards zugreifen.
Amazon Keyspaces CDC basiert auf den folgenden Prinzipien, auf die Sie sich beim Erstellen Ihrer Anwendung verlassen können:
+ Jeder Mutationsdatensatz auf Zeilenebene erscheint genau einmal im CDC-Stream.
+ Wenn Sie Shards in der Reihenfolge ihrer Abstammung konsumieren, erscheint jeder Mutationsdatensatz auf Zeilenebene in derselben Reihenfolge wie die tatsächliche Reihenfolge der Mutationen auf dem Primärschlüssel.
**Topics**
+ [Datenaufbewahrung](#CDC_how-it-works-data-retention)
+ [Ablauf von TTL-Daten](#CDC_how-it-works-ttl)
+ [Batch-Vorgänge](#CDC_how-it-works-batch-operations)
+ [Statische Spalten](#CDC_how-it-works-static)
+ [Verschlüsselung im Ruhezustand](#CDC_how-it-works-encryption)
+ [Multi-Region-Replikation](#CDC_how-it-works-mrr)
+ [Integration mit Diensten AWS](#howitworks_integration)
## So funktioniert die Datenspeicherung für CDC-Streams in Amazon Keyspaces
Amazon Keyspaces bewahrt die Datensätze im CDC-Stream für einen Zeitraum von 24 Stunden auf. Sie können die Aufbewahrungsfrist nicht ändern. Wenn Sie CDC für eine Tabelle deaktivieren, sind die Daten im Stream weiterhin 24 Stunden lang lesbar. Nach dieser Zeit laufen die Daten ab und die Datensätze werden automatisch gelöscht.
## So funktioniert der Ablauf von Time to Live (TTL) -Daten mit CDC-Streams in Amazon Keyspaces
Amazon Keyspaces zeigt die Ablaufzeit sowohl auf column/cell Ebene als auch auf Zeilenebene in einem Metadatenfeld an, das `expirationTime` in den CDC-Änderungsdatensätzen aufgerufen wird. Wenn Amazon Keyspaces TTL den Ablauf einer Zelle feststellt, erstellt CDC einen neuen Änderungsdatensatz, der TTL als Ursprung der Änderung anzeigt. Weitere Informationen zu TTL finden Sie unter [Daten mit Time to Live (TTL) für Amazon Keyspaces (für Apache Cassandra) ablaufen lassen](TTL.md).
## So funktionieren Batch-Operationen für CDC-Streams in Amazon Keyspaces
Batch-Operationen sind intern in einzelne Änderungen auf Zeilenebene unterteilt. Amazon Keyspaces speichert alle Datensätze innerhalb von CDC-Streams auf Zeilenebene, auch wenn die Änderung in einem Batch-Vorgang erfolgte. Amazon Keyspaces behält die Reihenfolge der Datensätze innerhalb des CDC-Streams in derselben Reihenfolge bei wie die Reihenfolge der Mutationen, die auf Zeilenebene oder auf dem Primärschlüssel aufgetreten sind.
## So funktionieren statische Spalten in CDC-Streams in Amazon Keyspaces
Statische Spaltenwerte werden von allen Zeilen in einer Partition in Cassandra gemeinsam genutzt. Aufgrund dieses Verhaltens erfasst Amazon Keyspaces alle Aktualisierungen einer statischen Spalte als separaten Datensatz im CDC-Stream. Die folgenden Beispiele fassen das Verhalten statischer Spaltenmutationen zusammen:
+ Wenn nur die statische Spalte aktualisiert wird, enthält der CDC-Stream eine Zeilenänderung für die statische Spalte als einzige Spalte in der Zeile.
+ Wenn eine Zeile aktualisiert wird, ohne dass die statische Spalte geändert wird, enthält der CDC-Stream eine Zeilenänderung, die alle Spalten außer der statischen Spalte enthält.
+ Wenn eine Zeile zusammen mit der statischen Spalte aktualisiert wird, enthält der CDC-Stream zwei separate Zeilenänderungen, eine für die statische Spalte und die andere für den Rest der Zeile.
## So funktioniert Verschlüsselung im Ruhezustand für CDC-Streams in Amazon Keyspaces
Um die ruhenden Daten im vom CDC georderten Protokoll zu verschlüsseln, verwendet Amazon Keyspaces denselben Verschlüsselungsschlüssel, der bereits für die Tabelle verwendet wird. Weitere Informationen zur Verschlüsselung im Ruhezustand finden Sie unter [Verschlüsselung im Ruhezustand in Amazon Keyspaces](EncryptionAtRest.md).
## So funktioniert die regionsübergreifende Replikation für CDC-Streams in Amazon Keyspaces
Sie können CDC-Streams für einzelne Replikate einer Tabelle mit mehreren Regionen aktivieren und deaktivieren, indem Sie entweder die `update-table` API oder den CQL-Befehl verwenden. `ALTER TABLE` Aufgrund der asynchronen Replikation und Konfliktlösung sind CDC-Streams für Tabellen mit mehreren Regionen nicht einheitlich. AWS-Regionen Daher können die Datensätze, die Amazon Keyspaces im Stream erfasst, in verschiedenen Regionen in einer anderen Reihenfolge angezeigt werden.
Weitere Informationen zur Replikation in mehreren Regionen finden Sie unter. [Multi-Region-Replikation für Amazon Keyspaces (für Apache Cassandra)](multiRegion-replication.md)
## CDC-Streams und Integration mit Diensten AWS
### So arbeiten Sie mit VPC-Endpunkten für CDC-Streams in Amazon Keyspaces
Sie können VPC-Endpunkte verwenden, um auf Amazon Keyspaces CDC-Streams zuzugreifen. Informationen zum Erstellen und Zugreifen auf VPC-Endpunkte für Streams finden Sie unter. [Verwendung von Amazon Keyspaces CDC-Streams mit Schnittstellen-VPC-Endpunkten](vpc-endpoints-streams.md)
### So CloudWatch funktioniert die Überwachung mit für CDC-Streams in Amazon Keyspaces
Sie können Amazon verwenden CloudWatch , um API-Aufrufe an den Amazon Keyspaces CDC-Endpunkt zu überwachen. Weitere Informationen zu den verfügbaren Metriken finden Sie unter. [Metriken für Amazon Keyspaces Change Data Capture (CDC)](metrics-dimensions.md#keyspaces-cdc-metrics)
### So CloudTrail funktioniert die Protokollierung mit für CDC-Streams in Amazon Keyspaces
Amazon Keyspaces CDC ist in einen Service integriert AWS CloudTrail, der eine Aufzeichnung der Aktionen eines Benutzers, einer Rolle oder eines AWS Dienstes in Amazon Keyspaces bereitstellt. CloudTrail erfasst Data Definition Language (DDL) -API-Aufrufe und Data Manipulation Language (DML) -API-Aufrufe für Amazon Keyspaces als Ereignisse. Zu den aufgezeichneten Aufrufen gehören Aufrufe von der Amazon Keyspaces-Konsole und programmatische Aufrufe der Amazon Keyspaces-API-Operationen.
Weitere Informationen zu den von erfassten CDC-Ereignissen finden Sie unter. CloudTrail [Protokollieren von Amazon Keyspaces-API-Aufrufen mit AWS CloudTrail](logging-using-cloudtrail.md)
### So funktioniert das Tagging für CDC-Streams in Amazon Keyspaces
Amazon Keyspaces CDC-Streams sind eine taggable Ressource. Sie können einen Stream taggen, wenn Sie eine Tabelle programmgesteuert mit CQL, dem SDK oder dem erstellen. AWS AWS CLI Sie können auch bestehende Streams taggen, Tags löschen oder Tags eines Streams anzeigen. Weitere Informationen finden Sie unter [Taggen Sie Schlüsselräume, Tabellen und Streams in Amazon Keyspaces](Tagging.Operations.md).
# So verwenden Sie Change Data Capture (CDC) -Streams in Amazon Keyspaces
**Topics**
+ [Konfigurieren von -Berechtigungen](configure-cdc-permissions.md)
+ [Greifen Sie auf CDC-Stream-Endpunkte zu](CDC_access-endpoints.md)
+ [Aktivieren Sie einen CDC-Stream für eine neue Tabelle](keyspaces-enable-cdc-new-table.md)
+ [Aktivieren Sie einen CDC-Stream für eine bestehende Tabelle](keyspaces-enable-cdc-alter-table.md)
+ [Deaktivieren Sie einen CDC-Stream](keyspaces-delete-cdc.md)
+ [CDC-Streams anzeigen](keyspaces-view-cdc.md)
+ [Greifen Sie auf CDC-Streams zu](keyspaces-records-cdc.md)
+ [Verwenden Sie KCL für die Verarbeitung von Streams](cdc_how-to-use-kcl.md)
# Berechtigungen für die Arbeit mit CDC-Streams in Amazon Keyspaces konfigurieren
Um CDC-Streams zu aktivieren, benötigt der Principal, z. B. ein IAM-Benutzer oder eine IAM-Rolle, die folgenden Berechtigungen.
Weitere Informationen zu finden Sie AWS Identity and Access Management unter. [AWS Identity and Access Management für Amazon Keyspaces](security-iam.md)
## Berechtigungen zum Aktivieren eines CDC-Streams für eine Tabelle
[Um einen CDC-Stream für eine Amazon Keyspaces-Tabelle zu aktivieren, benötigt der Principal zunächst Berechtigungen zum Erstellen oder Ändern einer Tabelle und zweitens die Berechtigungen zum Erstellen der serviceverknüpften Rolle AWSService RoleForAmazonKeyspaces CDC.](using-service-linked-roles-CDC-streams.md#service-linked-role-permissions-CDC-streams) Amazon Keyspaces verwendet die serviceverknüpfte Rolle, um in Ihrem Namen CloudWatch Metriken in Ihrem Konto zu veröffentlichen
Die folgende IAM-Richtlinie ist ein Beispiel dafür.
```
{
"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"
}
}
}
]
}
```
Um einen Stream zu deaktivieren, sind nur `ALTER TABLE` Berechtigungen erforderlich.
## Berechtigungen zum Anzeigen eines CDC-Streams
Um CDC-Streams anzuzeigen oder aufzulisten, benötigt der Principal Leseberechtigungen für den Systemschlüsselraum. Weitere Informationen finden Sie unter [`system_schema_mcs`](working-with-keyspaces.md#keyspace_system_schema_mcs).
Die folgende IAM-Richtlinie ist ein Beispiel dafür.
```
{
"Version":"2012-10-17",
"Statement":[
{
"Effect":"Allow",
"Action":"cassandra:Select",
"Resource":[
"arn:aws:cassandra:us-east-1:111122223333:/keyspace/system*"
]
}
]
}
```
Um CDC-Streams mit der oder der Amazon Keyspaces-API anzuzeigen AWS CLI oder aufzulisten, benötigt der Principal zusätzliche Berechtigungen für die Aktionen `cassandra:ListStreams` und. `cassandra:GetStream`
Die folgende IAM-Richtlinie ist ein Beispiel dafür.
```
{
"Version": "2012-10-17",
"Statement": [
{
"Effect": "Allow",
"Action": [
"cassandra:Select",
"cassandra:ListStreams",
"cassandra:GetStream"
],
"Resource": "*"
}
]
}
```
## Berechtigungen zum Lesen eines CDC-Streams
Zum Lesen von CDC-Streams benötigt der Principal die folgenden Berechtigungen.
```
{
"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"
]
}
]
}
```
## Berechtigungen zur Verarbeitung von Amazon Keyspaces CDC-Streams mit der Kinesis Client Library (KCL)
Um Amazon Keyspaces CDC-Streams mit KCL zu verarbeiten, benötigt der IAM-Principal die folgenden Berechtigungen.
+ `Amazon Keyspaces`— Schreibgeschützter Zugriff auf einen bestimmten Amazon Keyspaces CDC-Stream.
+ `DynamoDB`— Berechtigungen zum Erstellen von `shard lease` Tabellen, Lese- und Schreibzugriff auf die Tabellen und Lesezugriff auf den Index, wie für die KCL-Stream-Verarbeitung erforderlich.
+ `CloudWatch`— Berechtigungen zum Veröffentlichen von Metrikdaten aus Amazon Keyspaces CDC-Streams, die mit KCL verarbeitet werden, in den Namespace Ihrer KCL-Client-Anwendung in Ihrem Konto. CloudWatch Weitere Informationen zur Überwachung finden Sie unter [Überwachen der Kinesis-Clientbibliothek mit Amazon CloudWatch](https://docs.aws.amazon.com/streams/latest/dev/monitoring-with-kcl.html).
```
{
"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":"*"
}
]
}
```
# So greifen Sie in Amazon Keyspaces auf CDC-Stream-Endpunkte zu
Amazon Keyspaces unterhält separate [Endpunkte](programmatic.endpoints.md#global_endpoints) für keyspaces/tables und für CDC-Streams in jedem, AWS-Region wo Amazon Keyspaces verfügbar ist. Um auf einen CDC-Stream zuzugreifen, wählen Sie die Region der Tabelle aus und ersetzen Sie das Präfix durch das `cassandra` Präfix `cassandra-streams` im Endpunktnamen, wie im folgenden Beispiel gezeigt:
[\[See the AWS documentation website for more details\]](http://docs.aws.amazon.com/de_de/keyspaces/latest/devguide/CDC_access-endpoints.html)
Die folgende Tabelle enthält eine vollständige Liste der verfügbaren öffentlichen Endpunkte für. Amazon Keyspaces change data capture streams Amazon Keyspaces CDC streams unterstützt sowohl als auch IPv4 . IPv6 Alle öffentlichen Endpunkte sind beispielsweise Dual-Stack-Endpunkte`cassandra-streams.us-east-1.api.aws`, die für IPv4 und konfiguriert werden können. IPv6
[\[See the AWS documentation website for more details\]](http://docs.aws.amazon.com/de_de/keyspaces/latest/devguide/CDC_access-endpoints.html)
# Aktivieren Sie einen CDC-Stream, wenn Sie eine neue Tabelle in Amazon Keyspaces erstellen
Um beim Erstellen einer Tabelle einen CDC-Stream zu aktivieren, können Sie die `CREATE TABLE` Anweisung in CQL oder den Befehl mit dem `create-table` verwenden. AWS CLI
Für jede geänderte Zeile in der Tabelle kann Amazon Keyspaces basierend auf der `view_type` von `cdc_specification` Ihnen ausgewählten Zeile die folgenden Änderungen erfassen:
+ `NEW_AND_OLD_IMAGES`— beide Versionen der Zeile, vor und nach der Änderung. Dies ist die Standardeinstellung.
+ `NEW_IMAGE`— die Version der Zeile nach der Änderung.
+ `OLD_IMAGE`— die Version der Zeile vor der Änderung.
+ `KEYS_ONLY`— die Partitions- und Clusterschlüssel der Zeile, die geändert wurde.
Hinweise zum Taggen eines Streams finden Sie unter[Fügen Sie beim Erstellen einer Tabelle Tags zu einem neuen Stream hinzu](Tagging.Operations.new.table.stream.md).
**Anmerkung**
Amazon Keyspaces CDC erfordert das Vorhandensein einer serviceverknüpften Rolle (`AWSServiceRoleForAmazonKeyspacesCDC`), die in Ihrem Namen Metrikdaten aus Amazon Keyspaces CDC-Streams `"cloudwatch:namespace": "AWS/Cassandra"` in Ihrem CloudWatch Konto veröffentlicht. Diese Rolle wird automatisch für Sie erstellt. Weitere Informationen finden Sie unter [Rollen für Amazon Keyspaces CDC-Streams verwenden](using-service-linked-roles-CDC-streams.md).
------
#### [ Cassandra Query Language (CQL) ]
**Aktivieren Sie einen CDC-Stream, wenn Sie eine Tabelle mit CQL erstellen**
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. Um die Stream-Einstellungen zu bestätigen, können Sie die folgende Anweisung verwenden.
```
SELECT keyspace_name, table_name, cdc, custom_properties FROM system_schema_mcs.tables WHERE keyspace_name = 'mykeyspace' AND table_name = 'mytable';
```
Die Ausgabe dieser Anweisung sollte in etwa so aussehen.
```
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 ]
**Aktivieren Sie einen CDC-Stream, wenn Sie eine Tabelle mit dem erstellen AWS CLI**
1. Um einen Stream zu erstellen, können Sie die folgende Syntax verwenden.
```
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. Die Ausgabe dieses Befehls zeigt die `create-table` Standardantwort und sieht diesem Beispiel ähnlich.
```
{ "resourceArn": "arn:aws:cassandra:us-east-1:111122223333:/keyspace/mykeyspace/table/mytable" }
```
------
# Einen CDC-Stream für eine bestehende Tabelle in Amazon Keyspaces aktivieren
Um einen CDC-Stream für eine bestehende Tabelle zu aktivieren, können Sie die `ALTER TABLE` Anweisung in CQL, den `update-table` Befehl mit der AWS CLI oder die Konsole verwenden.
Für jede geänderte Zeile in der Tabelle kann Amazon Keyspaces basierend auf der `view_type` von `cdc_specification` Ihnen ausgewählten Zeile die folgenden Änderungen erfassen:
+ `NEW_AND_OLD_IMAGES`— beide Versionen der Zeile, vor und nach der Änderung. Dies ist die Standardeinstellung.
+ `NEW_IMAGE`— die Version der Zeile nach der Änderung.
+ `OLD_IMAGE`— die Version der Zeile vor der Änderung.
+ `KEYS_ONLY`— die Partitions- und Clusterschlüssel der Zeile, die geändert wurde.
Hinweise zum Taggen eines Streams finden Sie unter[Fügen Sie einem Stream neue Tags hinzu](Tagging.Operations.existing.stream.md).
**Anmerkung**
Amazon Keyspaces CDC erfordert das Vorhandensein einer serviceverknüpften Rolle (`AWSServiceRoleForAmazonKeyspacesCDC`), die in Ihrem Namen Metrikdaten aus Amazon Keyspaces CDC-Streams `"cloudwatch:namespace": "AWS/Cassandra"` in Ihrem CloudWatch Konto veröffentlicht. Diese Rolle wird automatisch für Sie erstellt. Weitere Informationen finden Sie unter [Rollen für Amazon Keyspaces CDC-Streams verwenden](using-service-linked-roles-CDC-streams.md).
------
#### [ Cassandra Query Language (CQL) ]
**Aktivieren Sie einen Stream (CDC-Stream) mit CQL**
Sie können verwenden`ALTER TABLE`, um einen Stream für eine bestehende Tabelle zu aktivieren.
1. Im folgenden Beispiel wird ein Stream erstellt, der nur Änderungen an den Partitions- und Clusterschlüsseln einer geänderten Zeile erfasst.
```
ALTER TABLE mykeyspace.mytable
WITH cdc = TRUE
AND CUSTOM_PROPERTIES={'cdc_specification': {'view_type': 'KEYS_ONLY'}};
```
1. Um die Stream-Einstellungen zu überprüfen, können Sie die folgende Anweisung verwenden.
```
SELECT keyspace_name, table_name, cdc, custom_properties FROM system_schema_mcs.tables WHERE keyspace_name = 'mykeyspace' AND table_name = 'mytable';
```
Die Ausgabe der Anweisung sieht in etwa so aus.
```
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 ]
**Erstellen Sie einen CDC-Stream mit AWS CLI**
1. Um einen Stream für eine bestehende Tabelle zu erstellen, können Sie die folgende Syntax verwenden.
```
aws keyspaces update-table \
--keyspace-name 'mykeyspace' \
--table-name 'mytable' \
--cdc-specification status=ENABLED,viewType=NEW_AND_OLD_IMAGES
```
1. Die Ausgabe dieses Befehls zeigt die `create-table` Standardantwort und sieht diesem Beispiel ähnlich.
```
{ "resourceArn": "arn:aws:cassandra:us-east-1:111122223333:/keyspace/mykeyspace/table/mytable" }
```
------
#### [ Console ]
**Aktivieren Sie einen CDC-Stream mit der Amazon Keyspaces-Konsole**
1. Melden Sie sich bei der AWS-Managementkonsole an und öffnen Sie die Amazon Keyspaces-Konsole zu [https://console.aws.amazon.com/keyspaces/Hause](https://console.aws.amazon.com/keyspaces/home).
1. Wählen Sie im Navigationsbereich **Tabellen** und dann eine Tabelle aus der Liste aus.
1. Wählen Sie die Registerkarte **Streams**.
1. Wählen Sie **Bearbeiten**, um einen Stream zu aktivieren.
1. Wählen Sie **Streams einschalten** aus.
1. Wählen Sie den **Ansichtstyp** des Streams. Verfügbar sind die nachfolgend aufgeführten Optionen. Beachten Sie, dass Sie den Ansichtstyp eines Streams nicht ändern können, nachdem er erstellt wurde.
+ **Neue und alte Bilder** — Amazon Keyspaces erfasst beide Versionen der Zeile vor und nach der Änderung. Dies ist die Standardeinstellung.
+ **Neues Bild** — Amazon Keyspaces erfasst nur die Version der Zeile nach der Änderung.
+ **Altes Bild** — Amazon Keyspaces erfasst nur die Version der Zeile vor der Änderung.
+ **Nur Primärschlüssel** — Amazon Keyspaces erfasst nur die Partitions- und Clusterschlüsselspalten der Zeile, die geändert wurde.
1. Um den Vorgang abzuschließen, wählen Sie Änderungen **speichern**.
------
# Deaktivieren Sie einen CDC-Stream in Amazon Keyspaces
Um einen CDC-Stream in einem Schlüsselraum zu deaktivieren, können Sie die `ALTER TABLE` Anweisung in CQL, den `update-table` Befehl mit der oder die Konsole verwenden. AWS CLI
------
#### [ Cassandra Query Language (CQL) ]
**Deaktivieren Sie einen Stream (CDC-Stream) mit CQL**
1. Um einen Stream zu deaktivieren, können Sie die folgende Anweisung verwenden.
```
ALTER TABLE mykeyspace.mytable
WITH cdc = FALSE;
```
1. Um zu bestätigen, dass der Stream deaktiviert ist, können Sie die folgende Anweisung verwenden.
```
SELECT keyspace_name, table_name, cdc, custom_properties FROM system_schema_mcs.tables WHERE keyspace_name = 'mykeyspace' AND table_name = 'mytable';
```
Die Ausgabe dieser Anweisung sieht in etwa so aus.
```
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 ]
**Deaktivieren Sie einen Stream (CDC-Stream) mit dem AWS CLI**
1. Um einen Stream zu deaktivieren, können Sie den folgenden Befehl verwenden.
```
aws keyspaces update-table \
--keyspace-name 'mykeyspace' \
--table-name 'mytable' \
--cdc-specification status=DISABLED
```
1. Die Ausgabe des Befehls sieht ähnlich aus wie in diesem Beispiel.
```
{
"keyspaceArn": "arn:aws:cassandra:us-east-1:111122223333:/keyspace/my_keyspace/",
"streamName": "my_stream"
}
```
------
#### [ Console ]
**Deaktivieren Sie einen Stream (CDC-Stream) mit der Amazon Keyspaces-Konsole**
1. Melden Sie sich bei der AWS-Managementkonsole an und öffnen Sie die Amazon Keyspaces-Konsole zu [https://console.aws.amazon.com/keyspaces/Hause](https://console.aws.amazon.com/keyspaces/home).
1. Wählen Sie im Navigationsbereich **Tabellen** und dann eine Tabelle aus der Liste aus.
1. Wählen Sie die Registerkarte **Streams**.
1. Wählen Sie **Bearbeiten** aus.
1. Deaktivieren Sie die Option **Streams einschalten**.
1. Wählen Sie **Änderungen speichern**, um den Stream zu deaktivieren.
------
# CDC-Streams in Amazon Keyspaces anzeigen
Um alle Streams im Schlüsselraum anzuzeigen oder aufzulisten, können Sie die Tabelle `system_schema_mcs.streams` im Systemschlüsselraum mit einer Anweisung in CQL abfragen oder die `list-stream` Befehle `get-stream` und mit der oder der AWS CLI Konsole verwenden.
Die erforderlichen Berechtigungen finden Sie unter [Berechtigungen für die Arbeit mit CDC-Streams in Amazon Keyspaces konfigurieren](configure-cdc-permissions.md).
------
#### [ Cassandra Query Language (CQL) ]
**CDC-Streams mit CQL anzeigen**
+ Um den CDC-Status Ihrer Tabelle zu überwachen, können Sie die folgende Anweisung verwenden.
```
SELECT custom_properties
FROM system_schema_mcs.tables
WHERE keyspace_name='my_keyspace' and table_name='my_table';
```
Die Ausgabe des Befehls sieht in etwa so aus.
```
...
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 ]
**Zeigen Sie CDC-Streams mit dem AWS CLI**
1. Dieses Beispiel zeigt, wie die Stream-Informationen für eine Tabelle angezeigt werden.
```
aws keyspaces get-table \
--keyspace-name 'my_keyspace' \
--table-name 'my_table'
```
Die Ausgabe des Befehls sieht wie folgt aus.
```
{
"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. Sie können alle Streams in Ihrem Konto in einer bestimmten Reihenfolge auflisten AWS-Region. Der folgende Befehl ist ein Beispiel dafür.
```
aws keyspacesstreams list-streams --region us-east-1
```
Die Ausgabe des Befehls könnte in etwa so aussehen.
```
{
"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. Sie können die CDC-Streams für einen bestimmten Schlüsselraum auch mithilfe der folgenden Parameter auflisten.
```
aws keyspacesstreams list-streams --keyspace-name ks_1 --region us-east-1
```
Die Ausgabe des Befehls sieht in etwa so aus.
```
{
"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. Sie können die CDC-Streams für eine bestimmte Tabelle auch mithilfe der folgenden Parameter auflisten.
```
aws keyspacesstreams list-streams --keyspace-name ks_1 --table-name t2 --region us-east-1
```
Die Ausgabe des Befehls sieht in etwa so aus.
```
{
"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 ]
**CDC-Streams in der Amazon Keyspaces-Konsole anzeigen**
1. Melden Sie sich bei der AWS-Managementkonsole an und öffnen Sie die Amazon Keyspaces-Konsole zu [https://console.aws.amazon.com/keyspaces/Hause](https://console.aws.amazon.com/keyspaces/home).
1. Wählen Sie im Navigationsbereich **Tabellen** und dann eine Tabelle aus der Liste aus.
1. Wählen Sie die Registerkarte **Streams**, um die Stream-Details zu überprüfen.
------
# Greifen Sie auf Datensätze in CDC-Streams in Amazon Keyspaces zu
Um auf die Datensätze in einem Stream zuzugreifen, verwenden Sie die [Amazon Keyspaces Streams API](https://docs.aws.amazon.com/keyspaces/latest/StreamsAPIReference/Welcome.html). Der folgende Abschnitt enthält Beispiele für den Zugriff auf Datensätze mithilfe von. AWS CLI
Die erforderlichen Berechtigungen finden Sie unter [Berechtigungen für die Arbeit mit CDC-Streams in Amazon Keyspaces konfigurieren](configure-cdc-permissions.md).
**Greifen Sie mit dem auf Datensätze in einem Stream zu AWS CLI**
1. Sie können die Amazon Keyspaces Streams-API verwenden, um auf die Änderungsdatensätze des Streams zuzugreifen. Weitere Informationen finden Sie unter [https://docs.aws.amazon.com/keyspaces/latest/StreamsAPIReference/Welcome.html](https://docs.aws.amazon.com/keyspaces/latest/StreamsAPIReference/Welcome.html) Um die Shards innerhalb des Streams abzurufen, können Sie die `get-stream` API verwenden, wie im folgenden Beispiel gezeigt.
```
aws keyspacesstreams get-stream \
--stream-arn 'arn:aws:cassandra:us-east-1:111122223333:/keyspace/mykeyspace/table/mytable/stream/STREAM_LABEL'
```
Es folgt ein Beispiel für die Ausgabe.
```
{
"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. Um Datensätze aus dem Stream abzurufen, rufen Sie zunächst einen Iterator auf, der Ihnen den Ausgangspunkt für den Zugriff auf Datensätze bietet. Dazu können Sie die Shards innerhalb des CDC-Streams verwenden, die von der API im vorherigen Schritt zurückgegeben wurden. Um den Iterator zu sammeln, können Sie die API verwenden. `get-shard-iterator` In diesem Beispiel verwenden Sie einen Iterator des Typs`TRIM_HORIZON`, der vom letzten gekürzten Punkt (oder Anfang) des Shards abruft.
```
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'
```
Die Ausgabe des Befehls sieht wie im folgenden Beispiel aus.
```
{
"ShardIterator": ""
}
```
1. Um die CDC-Datensätze mithilfe der `get-records` API abzurufen, können Sie den im letzten Schritt zurückgegebenen Iterator verwenden. Der folgende Befehl ist ein Beispiel dafür.
```
aws keyspacesstreams get-records \
--shard-iterator 'SHARD_ITERATOR' \
--limit 100
```
# Verwenden Sie die Kinesis Client Library (KCL), um Amazon Keyspaces-Streams zu verarbeiten
In diesem Thema wird beschrieben, wie Sie die Kinesis Client Library (KCL) verwenden, um Daten aus Amazon Keyspaces Change Data Capture (CDC) -Streams zu nutzen und zu verarbeiten.
Anstatt direkt mit der Amazon Keyspaces Streams API zu arbeiten, bietet die Arbeit mit der Kinesis Client Library (KCL) viele Vorteile, zum Beispiel:
+ Integriertes Shard-Lineage-Tracking und Iterator-Handling.
+ Automatischer Lastenausgleich zwischen Mitarbeitern.
+ Fehlertoleranz und Wiederherstellung nach Ausfällen von Mitarbeitern.
+ Checkpointing zur Nachverfolgung des Verarbeitungsfortschritts.
+ Anpassung an Änderungen der Stream-Kapazität.
+ Vereinfachtes verteiltes Rechnen für die Verarbeitung von CDC-Datensätzen.
Der folgende Abschnitt beschreibt, warum und wie die Kinesis Client Library (KCL) zur Verarbeitung von Streams verwendet wird, und bietet ein Beispiel für die Verarbeitung eines Amazon Keyspaces CDC-Streams mit der KCL.
Preisinformationen finden Sie unter [Amazon Keyspaces (für Apache Cassandra)](https://aws.amazon.com/keyspaces/pricing) — Preise.
## Was ist die Kinesis Client Library?
Die Kinesis Client Library (KCL) ist eine eigenständige Java-Softwarebibliothek, die entwickelt wurde, um den Prozess der Nutzung und Verarbeitung von Daten aus Streams zu vereinfachen. KCL bewältigt viele der komplexen Aufgaben im Zusammenhang mit verteilter Datenverarbeitung, sodass Sie sich bei der Verarbeitung von Stream-Daten auf die Implementierung Ihrer Geschäftslogik konzentrieren können. KCL verwaltet Aktivitäten wie den Lastenausgleich zwischen mehreren Workern, die Reaktion auf Worker-Ausfälle, Checkpoints verarbeiteter Datensätze und die Reaktion auf Änderungen der Anzahl der Shards im Stream.
Um Amazon Keyspaces CDC-Streams zu verarbeiten, können Sie die in der KCL enthaltenen Entwurfsmuster für die Arbeit mit Stream-Shards und Stream-Datensätzen verwenden. Die KCL vereinfacht die Codierung durch Bereitstellen nützlicher Abstraktionen oberhalb der Low-Level-Kinesis-Data-Streams-API. Weitere Informationen zur KCL finden Sie unter [Develop Consumers with KCL](https://docs.aws.amazon.com/kinesis/latest/dev/develop-kcl-consumers.html) im *Amazon Kinesis Data Streams* Developer Guide.
Um Anwendungen mit der KCL zu schreiben, verwenden Sie den Amazon Keyspaces Streams Kinesis Adapter. Der Kinesis-Adapter implementiert die Kinesis Data Streams-Schnittstelle, sodass Sie die KCL für die Nutzung und Verarbeitung von Datensätzen aus Amazon Keyspaces-Streams verwenden können. Anweisungen zur Einrichtung und Installation des Amazon Keyspaces Streams Kinesis-Adapters finden Sie im [GitHub](https://github.com/aws/keyspaces-streams-kinesis-adapter)Repository.
Das folgende Diagramm zeigt, wie diese Bibliotheken miteinander interagieren.
![\[Interaktion zwischen einer Client-Anwendung und Kinesis Data Streams, KCL, dem Amazon Keyspaces Streams Kinesis Adapter und Amazon Keyspaces bei der Verarbeitung von Amazon Keyspaces APIs CDC-Stream-Datensätzen.\]](http://docs.aws.amazon.com/de_de/keyspaces/latest/devguide/images/keyspaces-streams-kinesis-adapter.png)
KCL wird häufig aktualisiert, um neuere Versionen der zugrunde liegenden Bibliotheken, Sicherheitsverbesserungen und Bugfixes zu integrieren. Wir empfehlen Ihnen, die neueste Version von KCL zu verwenden, um bekannte Probleme zu vermeiden und von allen aktuellen Verbesserungen zu profitieren. [Die neueste KCL-Version finden Sie im KCL-Repository. GitHub ](https://github.com/awslabs/amazon-kinesis-client)
## KCL-Konzepte
Bevor Sie eine Verbraucheranwendung mithilfe von KCL implementieren, sollten Sie die folgenden Konzepte verstehen:
**KCL-Anwendung für Privatanwender**
Eine KCL-Verbraucheranwendung ist ein Programm, das Daten aus einem Amazon Keyspaces CDC-Stream verarbeitet. Die KCL fungiert als Vermittler zwischen Ihrem Verbraucheranwendungscode und dem Amazon Keyspaces CDC-Stream.
**Arbeiter**
Ein Worker ist eine Ausführungseinheit Ihrer KCL-Consumer-Anwendung, die Daten aus dem Amazon Keyspaces CDC-Stream verarbeitet. Ihre Anwendung kann mehrere Worker ausführen, die auf mehrere Instances verteilt sind.
**Prozessor aufnehmen**
Ein Record Processor ist die Logik in Ihrer Anwendung, die Daten aus einem Shard im Amazon Keyspaces CDC-Stream verarbeitet. Ein Datensatzprozessor wird von einem Worker für jeden von ihm verwalteten Shard instanziiert.
**Leasing**
Bei einem Leasing handelt es sich um die Verantwortung für die Verarbeitung eines Shards. Arbeiter koordinieren anhand von Leasingverträgen, welcher Arbeiter welchen Shard bearbeitet. KCL speichert Leasingdaten in einer Tabelle in Amazon DynamoDB.
**Checkpoint**
Ein Checkpoint ist eine Aufzeichnung der Position im Shard, bis zu der der Datensatzprozessor Datensätze erfolgreich verarbeitet hat. Checkpointing ermöglicht es Ihrer Anwendung, die Verarbeitung an der Stelle fortzusetzen, an der sie aufgehört hat, wenn ein Mitarbeiter ausfällt.
Mit dem Amazon Keyspaces Kinesis-Adapter können Sie mit der Entwicklung anhand der KCL-Schnittstelle beginnen, wobei die API-Aufrufe nahtlos an den Amazon Keyspaces-Stream-Endpunkt weitergeleitet werden. Eine Liste der verfügbaren Endpunkte finden Sie unter. [So greifen Sie in Amazon Keyspaces auf CDC-Stream-Endpunkte zu](CDC_access-endpoints.md)
Beim Start der Anwendung wird die KCL aufgerufen, einen Worker zu instanziieren. Sie müssen dem Worker Konfigurationsinformationen für die Anwendung zur Verfügung stellen, z. B. den Stream-Deskriptor und die AWS Anmeldeinformationen sowie den Namen einer von Ihnen angegebenen Datensatzprozessorklasse. Da der Code im Datensatzprozessor ausgeführt wird, erledigt der Worker die folgenden Aufgaben:
+ Stellt eine Verbindung mit dem Stream her
+ Listet die Shards innerhalb des Streams auf
+ Koordiniert Shard-Zuordnungen mit anderen Auftragnehmern (wenn vorhanden)
+ Instanziiert einen Datensatzverarbeiter für jeden Shard, der verwaltet wird
+ Ruft Datensätze aus dem Stream per Pull ab
+ Überträgt per Push Datensätze an den entsprechenden Datensatzverarbeiter
+ Verwendet Checkpoints für verarbeitete Datensätze
+ Gleicht Shard-Auftragnehmer-Zuordnungen aus, wenn die Auftragnehmer-Instance Änderungen zählt
+ Gleicht Shard-Worker-Zuordnungen aus, wenn Shards aufgeteilt werden
# Implementieren Sie eine KCL-Consumer-Anwendung für Amazon Keyspaces CDC-Streams
Dieses Thema enthält eine step-by-step Anleitung zur Implementierung einer KCL-Consumer-Anwendung zur Verarbeitung von Amazon Keyspaces CDC-Streams.
1. Voraussetzungen: Bevor Sie beginnen, stellen Sie sicher, dass Sie über Folgendes verfügen:
+ Eine Amazon Keyspaces-Tabelle mit einem CDC-Stream
+ Erforderliche IAM-Berechtigungen für den IAM-Principal, um auf den Amazon Keyspaces CDC-Stream zuzugreifen, DynamoDB-Tabellen für die KCL-Stream-Verarbeitung zu erstellen und darauf zuzugreifen, sowie Berechtigungen zum Veröffentlichen von Metriken in. CloudWatch Weitere Informationen und ein Richtlinienbeispiel finden Sie unter. [Berechtigungen zur Verarbeitung von Amazon Keyspaces CDC-Streams mit der Kinesis Client Library (KCL)](configure-cdc-permissions.md#cdc-permissions-kcl)
+ Stellen Sie sicher, dass in Ihrer lokalen Konfiguration gültige AWS Anmeldeinformationen eingerichtet sind. Weitere Informationen finden Sie unter [Speichern Sie die Zugriffstasten für den programmatischen Zugriff](aws.credentials.manage.md).
+ Java Development Kit (JDK) 8 oder höher
+ Die Anforderungen sind in der [Readme-Datei](https://github.com/aws/keyspaces-streams-kinesis-adapter) auf Github aufgeführt.
1. In diesem Schritt fügen Sie Ihrem Projekt die KCL-Abhängigkeit hinzu. Für Maven fügen Sie Ihrer pom.xml Folgendes hinzu:
```
software.amazon.kinesis
amazon-kinesis-client
3.1.0
software.amazon.keyspaces
keyspaces-streams-kinesis-adapter
1.0.0
```
**Anmerkung**
Suchen Sie im KCL-Repository immer nach der neuesten Version von [ GitHub KCL](https://github.com/awslabs/amazon-kinesis-client).
1. Erstellen Sie eine Factory-Klasse, die Datensatzprozessor-Instanzen erzeugt:
```
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. Erstellen Sie eine Record Factory, wie im folgenden Beispiel gezeigt.
```
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. In diesem Schritt erstellen Sie die zu konfigurierende Basisklasse KCLv3 und den Amazon Keyspaces-Adapter.
```
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. In diesem Schritt implementieren Sie die Datensatzprozessorklasse für Ihre Anwendung, um mit der Verarbeitung von Änderungsereignissen zu beginnen.
```
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);
}
}
```
## Best Practices
Folgen Sie diesen bewährten Methoden, wenn Sie KCL mit Amazon Keyspaces CDC-Streams verwenden:
**Fehlerbehandlung**
Implementieren Sie eine robuste Fehlerbehandlung in Ihrem Datensatzprozessor, um Ausnahmen ordnungsgemäß zu behandeln. Erwägen Sie die Implementierung einer Wiederholungslogik für vorübergehende Fehler.
**Häufigkeit des Checkpoints**
Sorgen Sie für eine ausgewogene Häufigkeit der Checkpoints, um doppelte Bearbeitungen zu minimieren und gleichzeitig eine angemessene Fortschrittsverfolgung zu gewährleisten. Zu häufiges Checkpointing kann sich negativ auf die Leistung auswirken, während ein zu seltenes Checkpointing zu mehr Wiederholungen führen kann, wenn ein Mitarbeiter ausfällt.
**Skalierung der Mitarbeiter**
Skalieren Sie die Anzahl der Worker basierend auf der Anzahl der Shards in Ihrem CDC-Stream. Ein guter Ausgangspunkt ist es, einen Worker pro Shard zu haben. Möglicherweise müssen Sie dies jedoch an Ihre Verarbeitungsanforderungen anpassen.
**Überwachung**
Verwenden Sie die von KCL bereitgestellten CloudWatch Kennzahlen, um den Zustand und die Leistung Ihrer Verbraucheranwendung zu überwachen. Zu den wichtigsten Kennzahlen gehören die Verarbeitungslatenz, das Alter der Checkpoints und die Anzahl der Leasingverträge.
**Testen**
Testen Sie Ihre Verbraucheranwendung gründlich, einschließlich Szenarien wie Worker-Ausfällen, Stream-Resharding und unterschiedlichen Lastbedingungen.
## Verwenden von KCL mit Nicht-Java-Sprachen
KCL ist zwar in erster Linie eine Java-Bibliothek, Sie können sie jedoch mit anderen Programmiersprachen über die verwenden. MultiLangDaemon Der MultiLangDaemon ist ein Java-basierter Daemon, der die Interaktion zwischen Ihrem Nicht-Java-Recordprozessor und der KCL verwaltet.
KCL bietet Unterstützung für die folgenden Sprachen:
+ Python
+ Ruby
+ Node.js
+ .NET
[Weitere Informationen zur Verwendung von KCL mit Nicht-Java-Sprachen finden Sie in der KCL-Dokumentation. MultiLangDaemon ](https://github.com/awslabs/amazon-kinesis-client/tree/master/amazon-kinesis-client-multilang)
## Fehlerbehebung
Dieser Abschnitt bietet Lösungen für häufig auftretende Probleme, die bei der Verwendung von KCL mit Amazon Keyspaces CDC-Streams auftreten können.
**Langsame Verarbeitung**
Wenn Ihre Verbraucheranwendung Daten langsam verarbeitet, sollten Sie Folgendes berücksichtigen:
+ Erhöhung der Anzahl der Worker-Instanzen
+ Optimieren Sie Ihre Logik zur Verarbeitung von Datensätzen
+ Überprüfung auf Engpässe in nachgeschalteten Systemen
**Doppelte Verarbeitung**
Wenn Sie feststellen, dass Datensätze doppelt verarbeitet werden, überprüfen Sie Ihre Checkpoint-Logik. Stellen Sie sicher, dass Sie nach erfolgreicher Verarbeitung von Datensätzen ein Checkpoint ausführen.
**Ausfälle von Arbeitern**
Wenn Mitarbeiter häufig ausfallen, überprüfen Sie Folgendes:
+ Ressourceneinschränkungen (CPU, Arbeitsspeicher)
+ Probleme mit der Netzwerkkonnektivität
+ Probleme mit den Berechtigungen
**Probleme mit der Leasetabelle**
Wenn Sie Probleme mit der KCL-Leasetabelle haben:
+ Überprüfen Sie, ob Ihre Anwendung über die entsprechenden Berechtigungen für den Zugriff auf die Amazon Keyspaces-Tabelle verfügt
+ Stellen Sie sicher, dass die Tabelle über einen ausreichenden bereitgestellten Durchsatz verfügt