Listagem de estilhaços

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

ListShardsAPI - Recomendado

O método recomendado para listar ou recuperar os fragmentos de um fluxo de dados é usar oListShardsAPI. O exemplo a seguir mostra como é possível obter uma lista de estilhaços em um stream 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, consulteListShards.

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 aListShardsAPI do, você pode usar oShardFilterparâmetro para filtrar a resposta da API. Você só pode especificar um filtro de cada vez.

Se você usar oShardFilterparâmetro ao invocar oListShardsAPI, oTypeÉ a propriedade necessária e deve ser especificada. Se você especificar oAT_TRIM_HORIZON,FROM_TRIM_HORIZON, ouAT_LATESTTipos, você não precisa especificar oShardIdou oTimestamppropriedades opcionais.

Se você especificar oAFTER_SHARD_IDTipo, você também deve fornecer o valor para o opcionalShardIdpropriedade. OShardIdpropriedade é idêntica em funcionalidade àExclusiveStartShardIdparâmetro doListShardsAPI. QuandoShardIdé especificada, a resposta inclui os fragmentos que começam com o fragmento cujo ID segue imediatamente oShardIdque você forneceu.

Se você especificar oAT_TIMESTAMPouFROM_TIMESTAMP_IDTipo, você também deve fornecer o valor para o opcionalTimestamppropriedade. Se você especificar oAT_TIMESTAMPtype, então todos os fragmentos que estavam abertos no carimbo de data/hora fornecido são retornados. Se você especificar oFROM_TIMESTAMPtipo, então todos os fragmentos a partir do carimbo de data/hora fornecido para TIP são retornados.

Importante

DescribeStreamSummaryeListShardAs APIs fornecem uma maneira mais escalável de recuperar informações sobre seus fluxos de dados. Mais especificamente, as cotas para oDescribeStreamA API pode causar limitação. Para obter mais informações, consulte Cotas e limites. Observe também queDescribeStreamAs cotas são compartilhadas entre todos os aplicativos que interagem com todos os streamings de dados do seuAWSconta. As cotas para oListShardsA API, por outro lado, é específica para um único stream de dados. Portanto, você não só obtém TPS mais alto com oListShardsAPI, mas a ação é melhor à medida que você cria mais fluxos de dados.

Recomendamos que você migre todos os seus produtores e consumidores que chamam deDescribeStreamAPI para invocar oDescribeStreamResumo e oListShardAPIs. Para identificar esses produtores e consumidores, recomendamos usar o Athena para analisarCloudTraillogs como 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
                

Recomendamos também que oAWSIntegrações do Lambda e Kinesis Data Firehose com o Kinesis Data Streams que invocam oDescribeStreamA API é reconfigurada para que as integrações invoquemDescribeStreamSummaryeListShards. Mais especificamente, paraAWSPor exemplo, você deve atualizar o mapeamento de origem do evento. Para o Kinesis Data Firehose, as permissões do IAM correspondentes devem ser atualizadas para que incluam oListShardsPermissão do IAM.

DescribeStreamAPI - Preterido

Importante

As informações abaixo descrevem uma maneira atualmente obsoleta de recuperar fragmentos de um fluxo de dados por meio doDescribeStreamAPI. No momento, é altamente recomendável usar oListShardsAPI 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 );