As traduções são geradas por tradução automática. Em caso de conflito entre o conteúdo da tradução e da versão original em inglês, a versão em inglês prevalecerá.
# Como usar fluxos de captura de dados de alteração (CDC) no Amazon Keyspaces
**Topics**
+ [Configurar permissões do](configure-cdc-permissions.md)
+ [Acesse os endpoints de fluxo do CDC](CDC_access-endpoints.md)
+ [Ativar um fluxo CDC para uma nova tabela](keyspaces-enable-cdc-new-table.md)
+ [Habilitar um fluxo CDC para uma tabela existente](keyspaces-enable-cdc-alter-table.md)
+ [Desativar um fluxo CDC](keyspaces-delete-cdc.md)
+ [Veja os streams do CDC](keyspaces-view-cdc.md)
+ [Acesse fluxos do CDC](keyspaces-records-cdc.md)
+ [Use o KCL para processar fluxos](cdc_how-to-use-kcl.md)
# Configure permissões para trabalhar com fluxos do CDC no Amazon Keyspaces
Para habilitar fluxos do CDC, o diretor, por exemplo, um usuário ou uma função do IAM, precisa das seguintes permissões.
Para obter mais informações sobre AWS Identity and Access Management, consulte[AWS Identity and Access Management para Amazon Keyspaces](security-iam.md).
## Permissões para habilitar um fluxo CDC para uma tabela
[Para habilitar um stream de CDC para uma tabela do Amazon Keyspaces, o principal primeiro precisa de permissões para criar ou alterar uma tabela e, em segundo lugar, de permissões para criar a função CDC vinculada AWSService RoleForAmazonKeyspaces ao serviço.](using-service-linked-roles-CDC-streams.md#service-linked-role-permissions-CDC-streams) O Amazon Keyspaces usa a função vinculada ao serviço para publicar CloudWatch métricas em sua conta em seu nome.
A política do IAM a seguir é um exemplo disso.
```
{
"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"
}
}
}
]
}
```
Para desativar um stream, somente `ALTER TABLE` permissões são necessárias.
## Permissões para visualizar um stream do CDC
Para visualizar ou listar fluxos do CDC, o diretor precisa de permissões de leitura para o espaço de teclas do sistema. Para obter mais informações, consulte [`system_schema_mcs`](working-with-keyspaces.md#keyspace_system_schema_mcs).
A política do IAM a seguir é um exemplo disso.
```
{
"Version":"2012-10-17",
"Statement":[
{
"Effect":"Allow",
"Action":"cassandra:Select",
"Resource":[
"arn:aws:cassandra:us-east-1:111122223333:/keyspace/system*"
]
}
]
}
```
Para visualizar ou listar streams do CDC com a API AWS CLI ou Amazon Keyspaces, o diretor precisa de permissões adicionais para as ações e. `cassandra:ListStreams` `cassandra:GetStream`
A política do IAM a seguir é um exemplo disso.
```
{
"Version": "2012-10-17",
"Statement": [
{
"Effect": "Allow",
"Action": [
"cassandra:Select",
"cassandra:ListStreams",
"cassandra:GetStream"
],
"Resource": "*"
}
]
}
```
## Permissões para ler um fluxo do CDC
Para ler fluxos do CDC, o diretor precisa das seguintes permissões.
```
{
"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"
]
}
]
}
```
## Permissões para processar streams CDC do Amazon Keyspaces com a Kinesis Client Library (KCL)
Para processar streams CDC do Amazon Keyspaces com a KCL, o diretor do IAM precisa das seguintes permissões.
+ `Amazon Keyspaces`— Acesso somente para leitura a um stream CDC especificado do Amazon Keyspaces.
+ `DynamoDB`— Permissões para criar `shard lease` tabelas, acesso de leitura e gravação às tabelas e acesso de leitura ao índice, conforme necessário para o processamento do fluxo KCL.
+ `CloudWatch`— Permissões para publicar dados métricos do processamento de streams CDC do Amazon Keyspaces com KCL no namespace do seu aplicativo cliente KCL em sua conta. CloudWatch Para obter mais informações sobre monitoramento, consulte [Monitorar a biblioteca cliente do Kinesis com a 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":"*"
}
]
}
```
# Como acessar endpoints de stream do CDC no Amazon Keyspaces
O Amazon Keyspaces mantém [endpoints](programmatic.endpoints.md#global_endpoints) separados para keyspaces/tables e para fluxos de CDC em cada um onde o Região da AWS Amazon Keyspaces está disponível. Para acessar um fluxo CDC, selecione a região da tabela e substitua o `cassandra` prefixo pelo nome do endpoint, conforme mostrado no exemplo a seguir: `cassandra-streams`
[\[See the AWS documentation website for more details\]](http://docs.aws.amazon.com/pt_br/keyspaces/latest/devguide/CDC_access-endpoints.html)
A tabela a seguir contém uma lista completa dos endpoints públicos disponíveis para Amazon Keyspaces change data capture streams. Amazon Keyspaces CDC streams suporta ambos IPv4 IPv6 e. Todos os endpoints públicos, por exemplo`cassandra-streams.us-east-1.api.aws`, são endpoints de pilha dupla que podem ser configurados para e. IPv4 IPv6
[\[See the AWS documentation website for more details\]](http://docs.aws.amazon.com/pt_br/keyspaces/latest/devguide/CDC_access-endpoints.html)
# Ative um stream do CDC ao criar uma nova tabela no Amazon Keyspaces
Para habilitar um stream CDC ao criar uma tabela, você pode usar a `CREATE TABLE` instrução em CQL ou o `create-table` comando com o. AWS CLI
Para cada linha alterada na tabela, o Amazon Keyspaces pode capturar as seguintes alterações com base nas `view_type` `cdc_specification` que você selecionar:
+ `NEW_AND_OLD_IMAGES`— ambas as versões da linha, antes e depois da alteração. Esse é o padrão.
+ `NEW_IMAGE`— a versão da linha após a alteração.
+ `OLD_IMAGE`— a versão da linha antes da alteração.
+ `KEYS_ONLY`— as chaves de partição e agrupamento da linha que foi alterada.
Para obter informações sobre como marcar um stream, consulte[Adicione tags a um novo stream ao criar uma tabela](Tagging.Operations.new.table.stream.md).
**nota**
O Amazon Keyspaces CDC exige a presença de uma função vinculada ao serviço (`AWSServiceRoleForAmazonKeyspacesCDC`) que publique dados métricos dos streams do Amazon Keyspaces CDC na sua conta em seu nome. `"cloudwatch:namespace": "AWS/Cassandra"` CloudWatch Esta função é criada automaticamente para você. Para obter mais informações, consulte [Usando funções para streams CDC do Amazon Keyspaces](using-service-linked-roles-CDC-streams.md).
------
#### [ Cassandra Query Language (CQL) ]
**Ative um fluxo CDC ao criar uma tabela com 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. Para confirmar as configurações de transmissão, você pode usar a seguinte declaração.
```
SELECT keyspace_name, table_name, cdc, custom_properties FROM system_schema_mcs.tables WHERE keyspace_name = 'mykeyspace' AND table_name = 'mytable';
```
A saída dessa declaração deve ser semelhante a essa.
```
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 ]
**Ative um fluxo CDC ao criar uma tabela com o AWS CLI**
1. Para criar um stream, você pode usar a seguinte sintaxe.
```
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. A saída desse comando mostra a `create-table` resposta padrão e é semelhante a este exemplo.
```
{ "resourceArn": "arn:aws:cassandra:us-east-1:111122223333:/keyspace/mykeyspace/table/mytable" }
```
------
# Habilite um stream CDC para uma tabela existente no Amazon Keyspaces
Para habilitar um stream CDC para uma tabela existente, você pode usar a `ALTER TABLE` instrução em CQL, o `update-table` comando com o. AWS CLI ou você pode usar o console.
Para cada linha alterada na tabela, o Amazon Keyspaces pode capturar as seguintes alterações com base nas `view_type` `cdc_specification` que você selecionar:
+ `NEW_AND_OLD_IMAGES`— ambas as versões da linha, antes e depois da alteração. Esse é o padrão.
+ `NEW_IMAGE`— a versão da linha após a alteração.
+ `OLD_IMAGE`— a versão da linha antes da alteração.
+ `KEYS_ONLY`— as chaves de partição e agrupamento da linha que foi alterada.
Para obter informações sobre como marcar um stream, consulte[Adicionar novas tags a um stream](Tagging.Operations.existing.stream.md).
**nota**
O Amazon Keyspaces CDC exige a presença de uma função vinculada ao serviço (`AWSServiceRoleForAmazonKeyspacesCDC`) que publique dados métricos dos streams do Amazon Keyspaces CDC na sua conta em seu nome. `"cloudwatch:namespace": "AWS/Cassandra"` CloudWatch Esta função é criada automaticamente para você. Para obter mais informações, consulte [Usando funções para streams CDC do Amazon Keyspaces](using-service-linked-roles-CDC-streams.md).
------
#### [ Cassandra Query Language (CQL) ]
**Ativar um fluxo (fluxo CDC) com CQL**
Você pode usar `ALTER TABLE` para habilitar um stream para uma tabela existente.
1. O exemplo a seguir cria um fluxo que captura somente as alterações na partição e nas chaves de agrupamento de uma linha alterada.
```
ALTER TABLE mykeyspace.mytable
WITH cdc = TRUE
AND CUSTOM_PROPERTIES={'cdc_specification': {'view_type': 'KEYS_ONLY'}};
```
1. Para verificar as configurações do stream, você pode usar a seguinte declaração.
```
SELECT keyspace_name, table_name, cdc, custom_properties FROM system_schema_mcs.tables WHERE keyspace_name = 'mykeyspace' AND table_name = 'mytable';
```
A saída da declaração é semelhante a essa.
```
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 ]
**Crie um fluxo CDC com o AWS CLI**
1. Para criar um fluxo para uma tabela existente, você pode usar a sintaxe a seguir.
```
aws keyspaces update-table \
--keyspace-name 'mykeyspace' \
--table-name 'mytable' \
--cdc-specification status=ENABLED,viewType=NEW_AND_OLD_IMAGES
```
1. A saída desse comando mostra a `create-table` resposta padrão e é semelhante a este exemplo.
```
{ "resourceArn": "arn:aws:cassandra:us-east-1:111122223333:/keyspace/mykeyspace/table/mytable" }
```
------
#### [ Console ]
**Habilite um stream do CDC com o console do Amazon Keyspaces**
1. [Faça login no Console de gerenciamento da AWS e abra o console do Amazon Keyspaces em casahttps://console.aws.amazon.com/keyspaces/.](https://console.aws.amazon.com/keyspaces/home)
1. No painel de navegação, escolha **Tabelas** e, em seguida, escolha uma tabela na lista.
1. Escolha a guia **Streams**.
1. Escolha **Editar** para ativar um stream.
1. Selecione **Ativar transmissões.**
1. Escolha o **tipo de exibição** do stream. As seguintes opções estão disponíveis: Observe que você não pode alterar o tipo de visualização de um stream depois que ele foi criado.
+ **Imagens novas e antigas** — o Amazon Keyspaces captura as duas versões da linha, antes e depois da alteração. Esse é o padrão.
+ **Nova imagem** — O Amazon Keyspaces captura somente a versão da linha após a alteração.
+ **Imagem antiga** — O Amazon Keyspaces captura somente a versão da linha antes da alteração.
+ **Somente chave primária** — O Amazon Keyspaces captura somente as colunas de partição e chave de agrupamento da linha que foi alterada.
1. Para finalizar, escolha **Salvar alterações**.
------
# Desativar um stream do CDC no Amazon Keyspaces
Para desativar um stream CDC em um keyspace, você pode usar a `ALTER TABLE` instrução em CQL, o `update-table` comando com o. ou o AWS CLI console.
------
#### [ Cassandra Query Language (CQL) ]
**Desativar um fluxo (fluxo CDC) com CQL**
1. Para desativar um stream, você pode usar a seguinte declaração.
```
ALTER TABLE mykeyspace.mytable
WITH cdc = FALSE;
```
1. Para confirmar se o stream está desativado, você pode usar a seguinte declaração.
```
SELECT keyspace_name, table_name, cdc, custom_properties FROM system_schema_mcs.tables WHERE keyspace_name = 'mykeyspace' AND table_name = 'mytable';
```
A saída dessa declaração é semelhante a essa.
```
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 ]
**Desative um fluxo (fluxo CDC) com o AWS CLI**
1. Para desativar um stream, você pode usar o comando a seguir.
```
aws keyspaces update-table \
--keyspace-name 'mykeyspace' \
--table-name 'mytable' \
--cdc-specification status=DISABLED
```
1. A saída do comando é semelhante a este exemplo.
```
{
"keyspaceArn": "arn:aws:cassandra:us-east-1:111122223333:/keyspace/my_keyspace/",
"streamName": "my_stream"
}
```
------
#### [ Console ]
**Desative um stream (stream CDC) com o console Amazon Keyspaces**
1. [Faça login no Console de gerenciamento da AWS e abra o console do Amazon Keyspaces em casahttps://console.aws.amazon.com/keyspaces/.](https://console.aws.amazon.com/keyspaces/home)
1. No painel de navegação, escolha **Tabelas** e, em seguida, escolha uma tabela na lista.
1. Escolha a guia **Streams**.
1. Escolha **Editar**.
1. Desmarque a opção **Ativar transmissões**.
1. Escolha **Salvar alterações** para desativar o stream.
------
# Veja os streams do CDC no Amazon Keyspaces
Para visualizar ou listar todos os fluxos no keyspace, você pode consultar a tabela `system_schema_mcs.streams` no keyspace do sistema usando uma instrução em CQL ou usar os `list-stream` comandos `get-stream` and com o AWS CLI, ou o console.
Para ver as permissões necessárias, consulte [Configure permissões para trabalhar com fluxos do CDC no Amazon Keyspaces](configure-cdc-permissions.md).
------
#### [ Cassandra Query Language (CQL) ]
**Veja os streams do CDC com o CQL**
+ Para monitorar o status do CDC da sua tabela, você pode usar a seguinte declaração.
```
SELECT custom_properties
FROM system_schema_mcs.tables
WHERE keyspace_name='my_keyspace' and table_name='my_table';
```
A saída do comando é semelhante a essa.
```
...
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 ]
**Veja os streams do CDC com o AWS CLI**
1. Este exemplo mostra como ver as informações de fluxo de uma tabela.
```
aws keyspaces get-table \
--keyspace-name 'my_keyspace' \
--table-name 'my_table'
```
A saída do comando tem a seguinte aparência.
```
{
"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. Você pode listar todos os streams em sua conta em um determinado Região da AWS local. O comando a seguir é um exemplo disso.
```
aws keyspacesstreams list-streams --region us-east-1
```
A saída do comando pode ser semelhante a essa.
```
{
"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. Você também pode listar os fluxos CDC para um determinado espaço de teclas usando os seguintes parâmetros.
```
aws keyspacesstreams list-streams --keyspace-name ks_1 --region us-east-1
```
A saída do comando é semelhante a essa.
```
{
"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. Você também pode listar os fluxos CDC de uma determinada tabela usando os parâmetros a seguir.
```
aws keyspacesstreams list-streams --keyspace-name ks_1 --table-name t2 --region us-east-1
```
A saída do comando é semelhante a essa.
```
{
"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 ]
**Veja os streams do CDC no console do Amazon Keyspaces**
1. [Faça login no Console de gerenciamento da AWS e abra o console do Amazon Keyspaces em casahttps://console.aws.amazon.com/keyspaces/.](https://console.aws.amazon.com/keyspaces/home)
1. No painel de navegação, escolha **Tabelas** e, em seguida, escolha uma tabela na lista.
1. Escolha a guia **Streams** para revisar os detalhes do stream.
------
# Acesse registros em fluxos do CDC no Amazon Keyspaces
Para acessar os registros em um stream, você usa a API [Amazon Keyspaces Streams](https://docs.aws.amazon.com/keyspaces/latest/StreamsAPIReference/Welcome.html). A seção a seguir contém exemplos de como acessar registros usando AWS CLI o.
Para ver as permissões necessárias, consulte [Configure permissões para trabalhar com fluxos do CDC no Amazon Keyspaces](configure-cdc-permissions.md).
**Acesse registros em um stream usando o AWS CLI**
1. Você pode usar a API Amazon Keyspaces Streams para acessar os registros de alterações do stream. Para obter mais informações, consulte [https://docs.aws.amazon.com/keyspaces/latest/StreamsAPIReference/Welcome.html](https://docs.aws.amazon.com/keyspaces/latest/StreamsAPIReference/Welcome.html) API Reference. Para recuperar os fragmentos no stream, você pode usar a `get-stream` API conforme mostrado no exemplo a seguir.
```
aws keyspacesstreams get-stream \
--stream-arn 'arn:aws:cassandra:us-east-1:111122223333:/keyspace/mykeyspace/table/mytable/stream/STREAM_LABEL'
```
A seguir, veja um exemplo de saída.
```
{
"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. Para recuperar registros do stream, você começa obtendo um iterador que fornece o ponto de partida para acessar os registros. Para fazer isso, você pode usar os fragmentos no stream do CDC retornado pela API na etapa anterior. Para reunir o iterador, você pode usar a `get-shard-iterator` API. Neste exemplo, você usa um iterador do tipo `TRIM_HORIZON` que recupera do último ponto recortado (ou início) do fragmento.
```
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'
```
A saída do comando se parece com o exemplo a seguir.
```
{
"ShardIterator": ""
}
```
1. Para recuperar os registros do CDC usando a `get-records` API, você pode usar o iterador retornado na última etapa. O comando a seguir é um exemplo disso.
```
aws keyspacesstreams get-records \
--shard-iterator 'SHARD_ITERATOR' \
--limit 100
```
# Use a Kinesis Client Library (KCL) para processar streams do Amazon Keyspaces
Este tópico descreve como usar a Kinesis Client Library (KCL) para consumir e processar dados dos streams de captura de dados de alteração (CDC) do Amazon Keyspaces.
Em vez de trabalhar diretamente com a API Amazon Keyspaces Streams, trabalhar com a Kinesis Client Library (KCL) oferece muitos benefícios, por exemplo:
+ Rastreamento de linhagem de fragmentos e manuseio de iteradores integrados.
+ Balanceamento automático de carga entre os trabalhadores.
+ Tolerância a falhas e recuperação de falhas de trabalhadores.
+ Ponto de verificação para acompanhar o progresso do processamento.
+ Adaptação às mudanças na capacidade do fluxo.
+ Computação distribuída simplificada para processamento de registros CDC.
A seção a seguir descreve por que e como usar a Kinesis Client Library (KCL) para processar streams e fornece um exemplo para processar um stream CDC do Amazon Keyspaces com a KCL.
Para obter informações sobre preços, consulte o [Amazon Keyspaces (para Apache Cassandra)](https://aws.amazon.com/keyspaces/pricing).
## O que é a Kinesis Client Library?
A Kinesis Client Library (KCL) é uma biblioteca de software Java independente projetada para simplificar o processo de consumo e processamento de dados de streams. A KCL lida com muitas das tarefas complexas associadas à computação distribuída, permitindo que você se concentre na implementação de sua lógica de negócios ao processar dados de fluxo. A KCL gerencia atividades como balanceamento de carga entre vários trabalhadores, resposta às falhas dos trabalhadores, verificação de registros processados e resposta às mudanças no número de fragmentos no fluxo.
Para processar streams CDC do Amazon Keyspaces, você pode usar os padrões de design encontrados na KCL para trabalhar com fragmentos de stream e registros de stream. A KCL simplifica a codificação fornecendo abstrações úteis acima da API de baixo nível do Kinesis Data Streams. Para obter mais informações sobre a KCL, consulte [Desenvolver consumidores com a KCL no Guia](https://docs.aws.amazon.com/kinesis/latest/dev/develop-kcl-consumers.html) do desenvolvedor do *Amazon Kinesis* Data Streams.
Para escrever aplicativos usando a KCL, você usa o Amazon Keyspaces Streams Kinesis Adapter. O Kinesis Adapter implementa a interface do Kinesis Data Streams para que você possa usar a KCL para consumir e processar registros dos streams do Amazon Keyspaces. Para obter instruções sobre como configurar e instalar o adaptador Kinesis de streams do Amazon Keyspaces, visite o repositório. [GitHub](https://github.com/aws/keyspaces-streams-kinesis-adapter)
O diagrama a seguir mostra como essas bibliotecas interagem entre si.
![\[Interação entre os aplicativos de um cliente e o Kinesis Data Streams, o KCL, o Amazon Keyspaces Streams Kinesis Adapter e o Amazon Keyspaces ao processar registros de stream CDC do APIs Amazon Keyspaces.\]](http://docs.aws.amazon.com/pt_br/keyspaces/latest/devguide/images/keyspaces-streams-kinesis-adapter.png)
A KCL é atualizada com frequência para incorporar as versões mais recentes das bibliotecas subjacentes, promover melhorias de segurança e realizar correções de bugs. É recomendável que você use a versão mais recente da KCL para evitar os problemas já conhecidos e se beneficiar de todas as melhorias mais recentes. Para encontrar a versão mais recente do KCL, consulte o repositório [KCL GitHub ](https://github.com/awslabs/amazon-kinesis-client).
## Conceitos da KCL
Antes de implementar um aplicativo de consumidor usando o KCL, você deve entender os seguintes conceitos:
**Aplicativo para consumidores KCL**
Um aplicativo de consumidor da KCL é um programa que processa dados de um stream CDC do Amazon Keyspaces. A KCL atua como intermediária entre o código do aplicativo do consumidor e o stream CDC do Amazon Keyspaces.
**Trabalhador**
Um trabalhador é uma unidade de execução do seu aplicativo consumidor da KCL que processa dados do stream CDC do Amazon Keyspaces. Seu aplicativo pode executar vários trabalhadores distribuídos em várias instâncias.
**Processador de gravação**
Um processador de registros é a lógica em seu aplicativo que processa dados de um fragmento no stream CDC do Amazon Keyspaces. Um processador de registros é instanciado por um trabalhador para cada fragmento que ele gerencia.
**Locação**
A concessão representa a responsabilidade pelo processamento de um fragmento. Os trabalhadores usam arrendamentos para coordenar qual trabalhador está processando qual fragmento. A KCL armazena dados de leasing em uma tabela no Amazon DynamoDB.
**Checkpoint**
Um ponto de verificação é um registro da posição no fragmento até a qual o processador de registros processou os registros com sucesso. O ponto de verificação permite que seu aplicativo retome o processamento de onde parou se um funcionário falhar.
Com o adaptador Amazon Keyspaces Kinesis instalado, você pode começar a desenvolver na interface KCL, com as chamadas de API direcionadas perfeitamente ao endpoint de stream do Amazon Keyspaces. Para obter uma lista de endpoints disponíveis, consulte [Como acessar endpoints de stream do CDC no Amazon Keyspaces](CDC_access-endpoints.md).
Quando a aplicação é iniciada, ela chama a KCL para instanciar um operador. Você deve fornecer ao trabalhador as informações de configuração do aplicativo, como o descritor e as AWS credenciais do fluxo, e o nome de uma classe de processador de registros fornecida por você. Como ele executa o código no processador de registro, o operador executa as seguintes tarefas:
+ Conecta-se ao stream
+ Enumera os fragmentos no fluxo
+ Coordena as associações do estilhaço com outros operadores (se houver)
+ Cria uma instância de um processador de registro para cada fragmento que gerencia
+ Extrai registros do fluxo
+ Envia os registros ao processador de registros correspondente
+ Registros processados pelos pontos de verificação
+ Equilibra as associações de estilhaço-operador quando a contagem de instância de operadores muda
+ Equilibra as associações de fragmento-operador quando os fragmentos se dividem
# Implemente um aplicativo de consumidor KCL para streams CDC do Amazon Keyspaces
Este tópico fornece um step-by-step guia para implementar um aplicativo de consumidor KCL para processar streams CDC do Amazon Keyspaces.
1. Pré-requisitos: Antes de começar, verifique se você tem:
+ Uma tabela do Amazon Keyspaces com um stream CDC
+ Permissões do IAM necessárias para que o diretor do IAM acesse o stream CDC do Amazon Keyspaces, crie e acesse tabelas do DynamoDB para processamento do stream KCL e permissões para publicar métricas. CloudWatch Para obter mais informações e um exemplo de política, consulte[Permissões para processar streams CDC do Amazon Keyspaces com a Kinesis Client Library (KCL)](configure-cdc-permissions.md#cdc-permissions-kcl).
+ Certifique-se de que AWS as credenciais válidas estejam configuradas em sua configuração local. Para obter mais informações, consulte [Armazene chaves de acesso para acesso programático](aws.credentials.manage.md).
+ Java Development Kit (JDK) 8 ou posterior
+ Requisitos listados no [Readme](https://github.com/aws/keyspaces-streams-kinesis-adapter) no Github.
1. Nesta etapa, você adiciona a dependência KCL ao seu projeto. Para o Maven, adicione o seguinte ao seu pom.xml:
```
software.amazon.kinesis
amazon-kinesis-client
3.1.0
software.amazon.keyspaces
keyspaces-streams-kinesis-adapter
1.0.0
```
**nota**
Sempre verifique a versão mais recente do KCL no repositório [KCL GitHub ](https://github.com/awslabs/amazon-kinesis-client).
1. Crie uma classe de fábrica que produza instâncias do processador de registros:
```
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. Crie uma fábrica de registros conforme mostrado no exemplo a seguir.
```
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. Nesta etapa, você cria a classe base a ser configurada KCLv3 e o adaptador 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. Nesta etapa, você implementa a classe do processador de registros para que seu aplicativo comece a processar eventos de alteração.
```
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);
}
}
```
## Práticas recomendadas
Siga estas melhores práticas ao usar a KCL com streams CDC do Amazon Keyspaces:
**Como tratar erros**
Implemente um tratamento robusto de erros em seu processador de registros para lidar com exceções sem problemas. Considere implementar a lógica de repetição para falhas transitórias.
**Frequência de checkpoint**
Equilibre a frequência dos pontos de verificação para minimizar o processamento duplicado e, ao mesmo tempo, garantir um rastreamento razoável do progresso. Pontos de verificação muito frequentes podem afetar o desempenho, enquanto pontos de verificação muito pouco frequentes podem levar a mais reprocessamento se um trabalhador falhar.
**Escalabilidade de trabalhadores**
Dimensione o número de trabalhadores com base no número de fragmentos em seu fluxo do CDC. Um bom ponto de partida é ter um trabalhador por fragmento, mas talvez seja necessário fazer ajustes com base nos requisitos de processamento.
**Monitoramento**
Use CloudWatch métricas fornecidas pela KCL para monitorar a integridade e o desempenho de seu aplicativo de consumo. As principais métricas incluem latência de processamento, idade do ponto de verificação e contagens de leasing.
**Teste**
Teste minuciosamente seu aplicativo de consumo, incluindo cenários como falhas de trabalho, refragmentação de streams e condições de carga variáveis.
## Usando KCL com linguagens não Java
Embora a KCL seja principalmente uma biblioteca Java, você pode usá-la com outras linguagens de programação por meio do MultiLangDaemon. MultiLangDaemon É um daemon baseado em Java que gerencia a interação entre seu processador de registros não Java e a KCL.
O KCL fornece suporte para os seguintes idiomas:
+ Python
+ Ruby
+ Node.js
+ .NET
Para obter mais informações sobre o uso do KCL com linguagens não Java, consulte a documentação do [ MultiLangDaemon KCL](https://github.com/awslabs/amazon-kinesis-client/tree/master/amazon-kinesis-client-multilang).
## Solução de problemas
Esta seção fornece soluções para problemas comuns que você pode encontrar ao usar a KCL com streams CDC do Amazon Keyspaces.
**Processamento lento**
Se seu aplicativo de consumidor estiver processando registros lentamente, considere:
+ Aumentar o número de instâncias de trabalho
+ Otimizando sua lógica de processamento de registros
+ Verificação de gargalos em sistemas a jusante
**Processamento duplicado**
Se você estiver vendo um processamento duplicado de registros, verifique sua lógica de ponto de verificação. Verifique se você está no ponto de verificação após processar os registros com sucesso.
**Falhas de trabalhadores**
Se os trabalhadores falharem com frequência, verifique:
+ Restrições de recursos (CPU, memória)
+ Problemas de conectividade de rede
+ Problemas com permissões
**Problemas com a tabela de locação**
Se você estiver enfrentando problemas com a tabela de leasing da KCL:
+ Verifique se seu aplicativo tem as permissões apropriadas para acessar a tabela Amazon Keyspaces
+ Verifique se a tabela tem taxa de transferência provisionada suficiente