列表分片 - Amazon Kinesis Data Streams
Amazon Web Services 文档中描述的 Amazon Web Services 服务或功能可能因区域而异。要查看适用于中国区域的差异,请参阅 中国的 Amazon Web Services 服务入门 (PDF)

本文属于机器翻译版本。若本译文内容与英语原文存在差异,则一律以英文原文为准。

列表分片

一个数据流可以有一个或多个分片。有两种方法可以列出(或检索)数据流中的分片。

ListShards API-推荐

从数据流中列出或检索分片的推荐方法是使用 API。ListShards以下示例说明如何获取数据流中的分片列表。有关本示例中使用的主操作的完整说明以及您可以为该操作设置的所有参数,请参阅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()); } } }

要运行上一个代码示例,您可以使用类似于下文的 POM 文件。

<?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>

通过 ListShards API,您可以使用ShardFilter参数筛选出 API 的响应。一次只能指定一个筛选条件。

如果您在调用 ListShards API 时使用ShardFilter参数,则Type为必填属性,必须指定。如果指定 AT_TRIM_HORIZONFROM_TRIM_HORIZONAT_LATEST 类型,则无需指定 ShardIdTimestamp 可选属性。

如果指定 AFTER_SHARD_ID 类型,则还必须提供可选 ShardId 属性的值。该ShardId属性的功能与 ListShards API 的ExclusiveStartShardId参数相同。指定 ShardId 属性后,响应将包括多个分片,其中开头的分片 ID 紧随您提供的 ShardId

如果指定 AT_TIMESTAMPFROM_TIMESTAMP_ID 类型,则还必须提供可选 Timestamp 属性的值。如果指定 AT_TIMESTAMP 类型,则返回在提供的时间戳上打开的所有分片。如果指定 FROM_TIMESTAMP 类型,则返回从提供的时间戳开始到 TIP 的所有分片。

重要

DescribeStreamSummaryListShard API 提供了一种更具可扩展性的方式来检索有关您的数据流的信息。更具体地说, DescribeStream API 的配额可能会导致限制。有关更多信息,请参阅配额和限制:另请注意,与您 Amazon 账户中所有数据流交互的所有应用程序均共享 DescribeStream 限额。另一方面, ListShards API 的配额特定于单个数据流。因此,使用 ListShards API 不仅可以获得更高的 TPS,而且随着您创建更多数据流,操作可以更好地扩展。

我们建议您迁移所有调用 DescribeStream API 的生产者和使用者,改为调用 DescribeStreamSummary 和 ListShard API。为了识别这些生产者和使用者,我们建议使用 Athena 来 CloudTrail 解析日志,因为在 API 调用中捕获 KPL 和 KCL 的用户代理。

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

我们还建议重新配置调用 DescribeStream API 的 Amazon Lambda 和 Amazon Firehose 与 Kinesis Data Streams 的集成,以便集成改为调用 DescribeStreamSummaryListShards。具体而言,对于 Amazon Lambda,您必须更新事件源映射。对于 Amazon Firehose,必须更新相应的 IAM 权限,使其包含 ListShards IAM 权限。

DescribeStream API-已弃用

重要

以下信息描述了目前已弃用的通过 API 从数据流中检索分片的方法。 DescribeStream 目前强烈建议您使用 ListShards API 来检索构成数据流的分片。

describeStream 方法返回的响应对象使您能够检索有关组成流的分片的信息。要检索分片,请对此对象调用 getShards 方法。此方法可能不会通过一次调用返回流中的所有分片。在以下代码中,我们查看 getHasMoreShards 上的 getStreamDescription 方法以了解是否有未返回的其他分片。如果有,也就是说,如果此方法返回 true,则我们将继续循环调用 getShards,以将返回的新的分片批次添加到我们的分片列表中。循环将在 getHasMoreShards 返回 false 时退出;即,所有分片均已返回。请注意,getShards 不会返回处于 EXPIRED 状态的分片。有关分片状态(包括 EXPIRED 状态)的更多信息,请参阅 分片之后的数据路由、数据保留和分片状态

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 );