Listagem de estilhaços - Amazon Kinesis Data Streams

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á.

Listagem de estilhaços

Um fluxo de dados pode ter um ou mais fragmentos. Há dois métodos para listar (ou recuperar) os fragmentos de um fluxo de dados.

ListShards API - Recomendado

O método recomendado para listar ou recuperar os fragmentos de um stream de dados é usar a ListShardsAPI. O exemplo a seguir mostra como obter uma lista de fragmentos em um fluxo de dados. Para obter uma descrição completa da operação principal usada neste exemplo e de todos os parâmetros que você pode definir para a operação, consulte ListShards.

import software.amazon.awssdk.services.kinesis.KinesisAsyncClient; import software.amazon.awssdk.services.kinesis.model.ListShardsRequest; import software.amazon.awssdk.services.kinesis.model.ListShardsResponse; import java.util.concurrent.TimeUnit; public class ShardSample { public static void main(String[] args) { KinesisAsyncClient client = KinesisAsyncClient.builder().build(); ListShardsRequest request = ListShardsRequest .builder().streamName("myFirstStream") .build(); try { ListShardsResponse response = client.listShards(request).get(5000, TimeUnit.MILLISECONDS); System.out.println(response.toString()); } catch (Exception e) { System.out.println(e.getMessage()); } } }

Para executar o exemplo de código anterior, você pode usar um arquivo POM, como o seguinte.

<?xml version="1.0" encoding="UTF-8"?> <project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd"> <modelVersion>4.0.0</modelVersion> <groupId>kinesis.data.streams.samples</groupId> <artifactId>shards</artifactId> <version>1.0-SNAPSHOT</version> <build> <plugins> <plugin> <groupId>org.apache.maven.plugins</groupId> <artifactId>maven-compiler-plugin</artifactId> <configuration> <source>8</source> <target>8</target> </configuration> </plugin> </plugins> </build> <dependencies> <dependency> <groupId>software.amazon.awssdk</groupId> <artifactId>kinesis</artifactId> <version>2.0.0</version> </dependency> </dependencies> </project>

Com a ListShards API, você pode usar o ShardFilterparâmetro para filtrar a resposta da API. Só é possível especificar um filtro de cada vez.

Se você usar o ShardFilter parâmetro ao invocar a ListShards API, essa Type é a propriedade necessária e deve ser especificada. Se você especificar os tipos AT_TRIM_HORIZON, FROM_TRIM_HORIZON ou AT_LATEST, não precisará especificar as propriedades opcionais ShardId e Timestamp.

Se você especificar o tipo AFTER_SHARD_ID, também deverá fornecer o valor para a propriedade opcional ShardId. A ShardId propriedade tem funcionalidade idêntica ao ExclusiveStartShardId parâmetro da ListShards API. Quando a propriedade ShardId é especificada, a resposta inclui os fragmentos a partir daquele cuja ID segue imediatamente a ShardId fornecida.

Se você especificar o tipo AT_TIMESTAMP ou FROM_TIMESTAMP_ID, também deverá fornecer o valor para a propriedade opcional Timestamp. Se você especificar o tipo AT_TIMESTAMP, todos os fragmentos abertos no timestamp fornecido serão retornados. Se você especificar o tipo FROM_TIMESTAMP, todos os fragmentos do timestamp fornecido até a extremidade serão retornados.

Importante

As APIs DescribeStreamSummary e ListShard fornecem uma maneira mais escalável de recuperar informações dos fluxos de dados. Mais especificamente, as cotas da DescribeStream API podem causar limitação. Para ter mais informações, consulte Cotas e limites. Observe também que as DescribeStream cotas são compartilhadas em todos os aplicativos que interagem com todos os fluxos de dados em sua AWS conta. As cotas da ListShards API, por outro lado, são específicas para um único fluxo de dados. Portanto, você não apenas obtém um TPS mais alto com a ListShards API, mas a ação é melhor dimensionada à medida que você cria mais fluxos de dados.

Recomendamos que você migre todos os produtores e consumidores que chamam a DescribeStream API para, em vez disso, invocar a DescribeStreamSummary e as ListShard APIs. Para identificar esses produtores e consumidores, recomendamos usar o Athena para analisar CloudTrail registros, pois os agentes de usuário para KPL e KCL são capturados nas chamadas de API.

SELECT useridentity.sessioncontext.sessionissuer.username, useridentity.arn,eventname,useragent, count(*) FROM cloudtrail_logs WHERE Eventname IN ('DescribeStream') AND eventtime BETWEEN '' AND '' GROUP BY useridentity.sessioncontext.sessionissuer.username,useridentity.arn,eventname,useragent ORDER BY count(*) DESC LIMIT 100

Também recomendamos que as integrações do AWS Lambda e do Amazon Firehose com o Kinesis Data Streams que invocam a API sejam reconfiguradas para que, em vez disso, DescribeStream as integrações invoquem e. DescribeStreamSummary ListShards Especificamente, para o AWS Lambda, você deve atualizar o mapeamento da fonte do evento. No Amazon Firehose, as permissões correspondentes do IAM precisam ser atualizadas para que incluam a permissão ListShards do IAM.

DescribeStream API - Obsoleto

Importante

As informações abaixo descrevem uma forma atualmente obsoleta de recuperar fragmentos de um fluxo de dados por meio da API. DescribeStream Atualmente, é altamente recomendável usar a API ListShards para recuperar os fragmentos que compõem o fluxo de dados.

O objeto de resposta retornado pelo método describeStream permite que você recupere informações sobre os estilhaços que compõem o stream. Para recuperar os estilhaços, chame o método getShards para esse objeto. Esse método pode não retornar todos os estilhaços do stream em uma única chamada. No código a seguir, verificamos o método getHasMoreShards em getStreamDescription para ver se há outros estilhaços que não foram retornados. Se houver, ou seja, se esse método retornar true, continuaremos chamando getShards em um loop, adicionando cada novo lote de estilhaços retornados à nossa lista de estilhaços. O loop é encerrado quando getHasMoreShards retorna false, ou seja, todos os estilhaços tiverem sido retornados. Observe que getShards não retorna estilhaços que estão no estado EXPIRED. Para obter mais informações sobre estados de estilhaço, incluindo o estado EXPIRED, consulte Roteamento de dados, persistência de dados e estado do estilhaço após um reestilhaçamento.

DescribeStreamRequest describeStreamRequest = new DescribeStreamRequest(); describeStreamRequest.setStreamName( myStreamName ); List<Shard> shards = new ArrayList<>(); String exclusiveStartShardId = null; do { describeStreamRequest.setExclusiveStartShardId( exclusiveStartShardId ); DescribeStreamResult describeStreamResult = client.describeStream( describeStreamRequest ); shards.addAll( describeStreamResult.getStreamDescription().getShards() ); if (describeStreamResult.getStreamDescription().getHasMoreShards() && shards.size() > 0) { exclusiveStartShardId = shards.get(shards.size() - 1).getShardId(); } else { exclusiveStartShardId = null; } } while ( exclusiveStartShardId != null );