文档 AWS SDK 示例 GitHub 存储库中还有更多 [S AWS DK 示例](https://github.com/awsdocs/aws-doc-sdk-examples)。
本文属于机器翻译版本。若本译文内容与英语原文存在差异，则一律以英文原文为准。
# 使用 Kinesis 的代码示例 AWS SDKs
以下代码示例向您展示了如何将 Amazon Kinesis 与 AWS 软件开发套件 (SDK) 配合使用。
*基本功能*是向您展示如何在服务中执行基本操作的代码示例。
*操作*是大型程序的代码摘录，必须在上下文中运行。您可以通过操作了解如何调用单个服务函数，还可以通过函数相关场景的上下文查看操作。
**更多资源**
+ **[Kinesis 开发人员指南](https://docs.aws.amazon.com/streams/latest/dev/introduction.html)**——有关 Kinesis 的更多信息。
+ **[Kinesis API 参考](https://docs.aws.amazon.com/kinesis/latest/APIReference/Welcome.html)**——有关所有可用的 Kinesis 操作的详细信息。
+ **[AWS 开发者中心](https://aws.amazon.com/developer/code-examples/?awsf.sdk-code-examples-product=product%23kinesis)** — 您可以按类别或全文搜索筛选的代码示例。
+ **[AWS SDK 示例](https://github.com/awsdocs/aws-doc-sdk-examples)** — 包含首选语言完整代码的 GitHub 存储库。包括有关设置和运行代码的说明。
**Contents**
+ [基本功能](kinesis_code_examples_basics.md)
+ [了解基本功能](kinesis_example_kinesis_Scenario_GettingStarted_section.md)
+ [操作](kinesis_code_examples_actions.md)
+ [`AddTagsToStream`](kinesis_example_kinesis_AddTagsToStream_section.md)
+ [`CreateStream`](kinesis_example_kinesis_CreateStream_section.md)
+ [`DeleteStream`](kinesis_example_kinesis_DeleteStream_section.md)
+ [`DeregisterStreamConsumer`](kinesis_example_kinesis_DeregisterStreamConsumer_section.md)
+ [`DescribeStream`](kinesis_example_kinesis_DescribeStream_section.md)
+ [`GetRecords`](kinesis_example_kinesis_GetRecords_section.md)
+ [`GetShardIterator`](kinesis_example_kinesis_GetShardIterator_section.md)
+ [`ListStreamConsumers`](kinesis_example_kinesis_ListStreamConsumers_section.md)
+ [`ListStreams`](kinesis_example_kinesis_ListStreams_section.md)
+ [`ListTagsForStream`](kinesis_example_kinesis_ListTagsForStream_section.md)
+ [`PutRecord`](kinesis_example_kinesis_PutRecord_section.md)
+ [`PutRecords`](kinesis_example_kinesis_PutRecords_section.md)
+ [`RegisterStreamConsumer`](kinesis_example_kinesis_RegisterStreamConsumer_section.md)
+ [无服务器示例](kinesis_code_examples_serverless_examples.md)
+ [通过 Kinesis 触发器调用 Lambda 函数](kinesis_example_serverless_Kinesis_Lambda_section.md)
+ [通过 Kinesis 触发器报告 Lambda 函数批处理项目失败](kinesis_example_serverless_Kinesis_Lambda_batch_item_failures_section.md)
# 使用 Kinesis 的基本示例 AWS SDKs
以下代码示例展示了如何使用 Amazon Kinesis 的基础知识。 AWS SDKs
**Contents**
+ [了解基本功能](kinesis_example_kinesis_Scenario_GettingStarted_section.md)
+ [操作](kinesis_code_examples_actions.md)
+ [`AddTagsToStream`](kinesis_example_kinesis_AddTagsToStream_section.md)
+ [`CreateStream`](kinesis_example_kinesis_CreateStream_section.md)
+ [`DeleteStream`](kinesis_example_kinesis_DeleteStream_section.md)
+ [`DeregisterStreamConsumer`](kinesis_example_kinesis_DeregisterStreamConsumer_section.md)
+ [`DescribeStream`](kinesis_example_kinesis_DescribeStream_section.md)
+ [`GetRecords`](kinesis_example_kinesis_GetRecords_section.md)
+ [`GetShardIterator`](kinesis_example_kinesis_GetShardIterator_section.md)
+ [`ListStreamConsumers`](kinesis_example_kinesis_ListStreamConsumers_section.md)
+ [`ListStreams`](kinesis_example_kinesis_ListStreams_section.md)
+ [`ListTagsForStream`](kinesis_example_kinesis_ListTagsForStream_section.md)
+ [`PutRecord`](kinesis_example_kinesis_PutRecord_section.md)
+ [`PutRecords`](kinesis_example_kinesis_PutRecords_section.md)
+ [`RegisterStreamConsumer`](kinesis_example_kinesis_RegisterStreamConsumer_section.md)
# 使用 SDK 学习 Kinesis 的基础知识 AWS
以下代码示例展示了如何：
+ 创建流并在其中放置一条记录。
+ 创建分片迭代器。
+ 读取记录，然后清理函数。
------
#### [ SAP ABAP ]
**适用于 SAP ABAP 的 SDK**
还有更多相关信息 GitHub。在 [AWS 代码示例存储库](https://github.com/awsdocs/aws-doc-sdk-examples/tree/main/sap-abap/services/kns#code-examples)中查找完整示例，了解如何进行设置和运行。
```
DATA lo_stream_describe_result TYPE REF TO /aws1/cl_knsdescrstreamoutput.
DATA lo_stream_description TYPE REF TO /aws1/cl_knsstreamdescription.
DATA lo_sharditerator TYPE REF TO /aws1/cl_knsgetsharditerator01.
DATA lo_record_result TYPE REF TO /aws1/cl_knsputrecordoutput.
"Create stream."
TRY.
lo_kns->createstream(
iv_streamname = iv_stream_name
iv_shardcount = iv_shard_count ).
MESSAGE 'Stream created.' TYPE 'I'.
CATCH /aws1/cx_knsinvalidargumentex.
MESSAGE 'The specified argument was not valid.' TYPE 'E'.
CATCH /aws1/cx_knslimitexceededex.
MESSAGE 'The request processing has failed because of a limit exceeded exception.' TYPE 'E'.
CATCH /aws1/cx_knsresourceinuseex.
MESSAGE 'The request processing has failed because the resource is in use.' TYPE 'E'.
ENDTRY.
"Wait for stream to becomes active."
lo_stream_describe_result = lo_kns->describestream( iv_streamname = iv_stream_name ).
lo_stream_description = lo_stream_describe_result->get_streamdescription( ).
WHILE lo_stream_description->get_streamstatus( ) <> 'ACTIVE'.
IF sy-index = 30.
EXIT. "maximum 5 minutes"
ENDIF.
WAIT UP TO 10 SECONDS.
lo_stream_describe_result = lo_kns->describestream( iv_streamname = iv_stream_name ).
lo_stream_description = lo_stream_describe_result->get_streamdescription( ).
ENDWHILE.
"Create record."
TRY.
lo_record_result = lo_kns->putrecord(
iv_streamname = iv_stream_name
iv_data = iv_data
iv_partitionkey = iv_partition_key ).
MESSAGE 'Record created.' TYPE 'I'.
CATCH /aws1/cx_knsinvalidargumentex.
MESSAGE 'The specified argument was not valid.' TYPE 'E'.
CATCH /aws1/cx_knskmsaccessdeniedex.
MESSAGE 'You do not have permission to perform this AWS KMS action.' TYPE 'E'.
CATCH /aws1/cx_knskmsdisabledex.
MESSAGE 'KMS key used is disabled.' TYPE 'E'.
CATCH /aws1/cx_knskmsinvalidstateex.
MESSAGE 'KMS key used is in an invalid state. ' TYPE 'E'.
CATCH /aws1/cx_knskmsnotfoundex.
MESSAGE 'KMS key used is not found.' TYPE 'E'.
CATCH /aws1/cx_knskmsoptinrequired.
MESSAGE 'KMS key option is required.' TYPE 'E'.
CATCH /aws1/cx_knskmsthrottlingex.
MESSAGE 'The rate of requests to AWS KMS is exceeding the request quotas.' TYPE 'E'.
CATCH /aws1/cx_knsprovthruputexcdex.
MESSAGE 'The request rate for the stream is too high, or the requested data is too large for the available throughput.' TYPE 'E'.
CATCH /aws1/cx_knsresourcenotfoundex.
MESSAGE 'Resource being accessed is not found.' TYPE 'E'.
ENDTRY.
"Create a shard iterator in order to read the record."
TRY.
lo_sharditerator = lo_kns->getsharditerator(
iv_shardid = lo_record_result->get_shardid( )
iv_sharditeratortype = iv_sharditeratortype
iv_streamname = iv_stream_name ).
MESSAGE 'Shard iterator created.' TYPE 'I'.
CATCH /aws1/cx_knsinvalidargumentex.
MESSAGE 'The specified argument was not valid.' TYPE 'E'.
CATCH /aws1/cx_knsprovthruputexcdex.
MESSAGE 'The request rate for the stream is too high, or the requested data is too large for the available throughput.' TYPE 'E'.
CATCH /aws1/cx_sgmresourcenotfound.
MESSAGE 'Resource being accessed is not found.' TYPE 'E'.
ENDTRY.
"Read the record."
TRY.
oo_result = lo_kns->getrecords( " oo_result is returned for testing purposes. "
iv_sharditerator = lo_sharditerator->get_sharditerator( ) ).
MESSAGE 'Shard iterator created.' TYPE 'I'.
CATCH /aws1/cx_knsexpirediteratorex.
MESSAGE 'Iterator expired.' TYPE 'E'.
CATCH /aws1/cx_knsinvalidargumentex.
MESSAGE 'The specified argument was not valid.' TYPE 'E'.
CATCH /aws1/cx_knskmsaccessdeniedex.
MESSAGE 'You do not have permission to perform this AWS KMS action.' TYPE 'E'.
CATCH /aws1/cx_knskmsdisabledex.
MESSAGE 'KMS key used is disabled.' TYPE 'E'.
CATCH /aws1/cx_knskmsinvalidstateex.
MESSAGE 'KMS key used is in an invalid state. ' TYPE 'E'.
CATCH /aws1/cx_knskmsnotfoundex.
MESSAGE 'KMS key used is not found.' TYPE 'E'.
CATCH /aws1/cx_knskmsoptinrequired.
MESSAGE 'KMS key option is required.' TYPE 'E'.
CATCH /aws1/cx_knskmsthrottlingex.
MESSAGE 'The rate of requests to AWS KMS is exceeding the request quotas.' TYPE 'E'.
CATCH /aws1/cx_knsprovthruputexcdex.
MESSAGE 'The request rate for the stream is too high, or the requested data is too large for the available throughput.' TYPE 'E'.
CATCH /aws1/cx_knsresourcenotfoundex.
MESSAGE 'Resource being accessed is not found.' TYPE 'E'.
ENDTRY.
"Delete stream."
TRY.
lo_kns->deletestream(
iv_streamname = iv_stream_name ).
MESSAGE 'Stream deleted.' TYPE 'I'.
CATCH /aws1/cx_knslimitexceededex.
MESSAGE 'The request processing has failed because of a limit exceeded exception.' TYPE 'E'.
CATCH /aws1/cx_knsresourceinuseex.
MESSAGE 'The request processing has failed because the resource is in use.' TYPE 'E'.
ENDTRY.
```
+ 有关 API 详细信息，请参阅《AWS SDK for SAP ABAP API Reference》**中的以下主题。
+ [CreateStream](https://docs.aws.amazon.com/sdk-for-sap-abap/v1/api/latest/index.html)
+ [DeleteStream](https://docs.aws.amazon.com/sdk-for-sap-abap/v1/api/latest/index.html)
+ [GetRecords](https://docs.aws.amazon.com/sdk-for-sap-abap/v1/api/latest/index.html)
+ [GetShardIterator](https://docs.aws.amazon.com/sdk-for-sap-abap/v1/api/latest/index.html)
+ [PutRecord](https://docs.aws.amazon.com/sdk-for-sap-abap/v1/api/latest/index.html)
------
# 使用 Kinesis 的操作 AWS SDKs
以下代码示例演示了如何使用执行各个 Kinesis 操作。 AWS SDKs每个示例都包含一个指向的链接 GitHub，您可以在其中找到有关设置和运行代码的说明。
以下示例仅包括最常用的操作。有关完整列表，请参阅 [Amazon Kinesis API 参考](https://docs.aws.amazon.com/kinesis/latest/APIReference/Welcome.html)。
**Topics**
+ [`AddTagsToStream`](kinesis_example_kinesis_AddTagsToStream_section.md)
+ [`CreateStream`](kinesis_example_kinesis_CreateStream_section.md)
+ [`DeleteStream`](kinesis_example_kinesis_DeleteStream_section.md)
+ [`DeregisterStreamConsumer`](kinesis_example_kinesis_DeregisterStreamConsumer_section.md)
+ [`DescribeStream`](kinesis_example_kinesis_DescribeStream_section.md)
+ [`GetRecords`](kinesis_example_kinesis_GetRecords_section.md)
+ [`GetShardIterator`](kinesis_example_kinesis_GetShardIterator_section.md)
+ [`ListStreamConsumers`](kinesis_example_kinesis_ListStreamConsumers_section.md)
+ [`ListStreams`](kinesis_example_kinesis_ListStreams_section.md)
+ [`ListTagsForStream`](kinesis_example_kinesis_ListTagsForStream_section.md)
+ [`PutRecord`](kinesis_example_kinesis_PutRecord_section.md)
+ [`PutRecords`](kinesis_example_kinesis_PutRecords_section.md)
+ [`RegisterStreamConsumer`](kinesis_example_kinesis_RegisterStreamConsumer_section.md)
# `AddTagsToStream`与 AWS SDK 或 CLI 配合使用
以下代码示例演示如何使用 `AddTagsToStream`。
------
#### [ .NET ]
**适用于 .NET 的 SDK**
还有更多相关信息 GitHub。在 [AWS 代码示例存储库](https://github.com/awsdocs/aws-doc-sdk-examples/tree/main/dotnetv3/Kinesis#code-examples)中查找完整示例，了解如何进行设置和运行。
```
using System;
using System.Collections.Generic;
using System.Threading.Tasks;
using Amazon.Kinesis;
using Amazon.Kinesis.Model;
///
/// This example shows how to apply key/value pairs to an Amazon Kinesis
/// stream.
///
public class TagStream
{
public static async Task Main()
{
IAmazonKinesis client = new AmazonKinesisClient();
string streamName = "AmazonKinesisStream";
var tags = new Dictionary
{
{ "Project", "Sample Kinesis Project" },
{ "Application", "Sample Kinesis App" },
};
var success = await ApplyTagsToStreamAsync(client, streamName, tags);
if (success)
{
Console.WriteLine($"Taggs successfully added to {streamName}.");
}
else
{
Console.WriteLine("Tags were not added to the stream.");
}
}
///
/// Applies the set of tags to the named Kinesis stream.
///
/// The initialized Kinesis client.
/// The name of the Kinesis stream to which
/// the tags will be attached.
/// A sictionary containing key/value pairs which
/// will be used to create the Kinesis tags.
/// A Boolean value which represents the success or failure
/// of AddTagsToStreamAsync.
public static async Task ApplyTagsToStreamAsync(
IAmazonKinesis client,
string streamName,
Dictionary tags)
{
var request = new AddTagsToStreamRequest
{
StreamName = streamName,
Tags = tags,
};
var response = await client.AddTagsToStreamAsync(request);
return response.HttpStatusCode == System.Net.HttpStatusCode.OK;
}
}
```
+ 有关 API 的详细信息，请参阅 *适用于 .NET 的 AWS SDK API 参考[AddTagsToStream](https://docs.aws.amazon.com/goto/DotNetSDKV3/kinesis-2013-12-02/AddTagsToStream)*中的。
------
#### [ CLI ]
**AWS CLI**
**为数据流添加标签**
以下 `add-tags-to-stream` 示例将键为 `samplekey` 且值为 `example` 的标签分配给指定流。
```
aws kinesis add-tags-to-stream \
--stream-name samplestream \
--tags samplekey=example
```
此命令不生成任何输出。
有关更多信息，请参阅《Amazon Kinesis Data Streams 开发人员指南》**中的[标记流](https://docs.aws.amazon.com/streams/latest/dev/tagging.html)。
+ 有关 API 的详细信息，请参阅*AWS CLI 命令参考[AddTagsToStream](https://awscli.amazonaws.com/v2/documentation/api/latest/reference/kinesis/add-tags-to-stream.html)*中的。
------
# `CreateStream`与 AWS SDK 或 CLI 配合使用
以下代码示例演示如何使用 `CreateStream`。
操作示例是大型程序的代码摘录，必须在上下文中运行。在以下代码示例中，您可以查看此操作的上下文：
+ [了解基本功能](kinesis_example_kinesis_Scenario_GettingStarted_section.md)
------
#### [ .NET ]
**适用于 .NET 的 SDK**
还有更多相关信息 GitHub。在 [AWS 代码示例存储库](https://github.com/awsdocs/aws-doc-sdk-examples/tree/main/dotnetv3/Kinesis#code-examples)中查找完整示例，了解如何进行设置和运行。
```
using System;
using System.Threading.Tasks;
using Amazon.Kinesis;
using Amazon.Kinesis.Model;
///
/// This example shows how to create a new Amazon Kinesis stream.
///
public class CreateStream
{
public static async Task Main()
{
IAmazonKinesis client = new AmazonKinesisClient();
string streamName = "AmazonKinesisStream";
int shardCount = 1;
var success = await CreateNewStreamAsync(client, streamName, shardCount);
if (success)
{
Console.WriteLine($"The stream, {streamName} successfully created.");
}
}
///
/// Creates a new Kinesis stream.
///
/// An initialized Kinesis client.
/// The name for the new stream.
/// The number of shards the new stream will
/// use. The throughput of the stream is a function of the number of
/// shards; more shards are required for greater provisioned
/// throughput.
/// A Boolean value indicating whether the stream was created.
public static async Task CreateNewStreamAsync(IAmazonKinesis client, string streamName, int shardCount)
{
var request = new CreateStreamRequest
{
StreamName = streamName,
ShardCount = shardCount,
};
var response = await client.CreateStreamAsync(request);
return response.HttpStatusCode == System.Net.HttpStatusCode.OK;
}
}
```
+ 有关 API 的详细信息，请参阅 *适用于 .NET 的 AWS SDK API 参考[CreateStream](https://docs.aws.amazon.com/goto/DotNetSDKV3/kinesis-2013-12-02/CreateStream)*中的。
------
#### [ CLI ]
**AWS CLI**
**创建数据流**
以下 `create-stream` 示例创建一个名为 samplestream 的数据流，其中包含 3 个分片。
```
aws kinesis create-stream \
--stream-name samplestream \
--shard-count 3
```
此命令不生成任何输出。
有关更多信息，请参阅《Amazon Kinesis Data Streams 开发人员指南》**中的[创建流](https://docs.aws.amazon.com/streams/latest/dev/kinesis-using-sdk-java-create-stream.html)。
+ 有关 API 的详细信息，请参阅*AWS CLI 命令参考[CreateStream](https://awscli.amazonaws.com/v2/documentation/api/latest/reference/kinesis/create-stream.html)*中的。
------
#### [ Java ]
**适用于 Java 的 SDK 2.x**
还有更多相关信息 GitHub。在 [AWS 代码示例存储库](https://github.com/awsdocs/aws-doc-sdk-examples/tree/main/javav2/example_code/kinesis#code-examples)中查找完整示例，了解如何进行设置和运行。
```
import software.amazon.awssdk.regions.Region;
import software.amazon.awssdk.services.kinesis.KinesisClient;
import software.amazon.awssdk.services.kinesis.model.CreateStreamRequest;
import software.amazon.awssdk.services.kinesis.model.KinesisException;
/**
* Before running this Java V2 code example, set up your development
* environment, including your credentials.
*
* For more information, see the following documentation topic:
*
* https://docs.aws.amazon.com/sdk-for-java/latest/developer-guide/get-started.html
*/
public class CreateDataStream {
public static void main(String[] args) {
final String usage = """
Usage:
Where:
streamName - The Amazon Kinesis data stream (for example, StockTradeStream).
""";
if (args.length != 1) {
System.out.println(usage);
System.exit(1);
}
String streamName = args[0];
Region region = Region.US_EAST_1;
KinesisClient kinesisClient = KinesisClient.builder()
.region(region)
.build();
createStream(kinesisClient, streamName);
System.out.println("Done");
kinesisClient.close();
}
public static void createStream(KinesisClient kinesisClient, String streamName) {
try {
CreateStreamRequest streamReq = CreateStreamRequest.builder()
.streamName(streamName)
.shardCount(1)
.build();
kinesisClient.createStream(streamReq);
} catch (KinesisException e) {
System.err.println(e.getMessage());
System.exit(1);
}
}
}
```
+ 有关 API 的详细信息，请参阅 *AWS SDK for Java 2.x API 参考[CreateStream](https://docs.aws.amazon.com/goto/SdkForJavaV2/kinesis-2013-12-02/CreateStream)*中的。
------
#### [ PowerShell ]
**适用于 PowerShell V4 的工具**
**示例 1：创建新流。默认情况下，此 cmdlet 不返回任何输出，因此添加-PassThru 开关以返回提供给-StreamName 参数的值以供日后使用。**
```
$streamName = New-KINStream -StreamName "mystream" -ShardCount 1 -PassThru
```
+ 有关 API 的详细信息，请参阅 *AWS Tools for PowerShell Cmdlet 参考 (V* 4) [CreateStream](https://docs.aws.amazon.com/powershell/v4/reference)中的。
**适用于 PowerShell V5 的工具**
**示例 1：创建新流。**
```
New-KINStream -StreamName "mystream" -ShardCount 1
```
+ 有关 API 的详细信息，请参阅 *AWS Tools for PowerShell Cmdlet 参考 (V* 5) [CreateStream](https://docs.aws.amazon.com/powershell/v5/reference)中的。
------
#### [ Python ]
**适用于 Python 的 SDK（Boto3）**
还有更多相关信息 GitHub。在 [AWS 代码示例存储库](https://github.com/awsdocs/aws-doc-sdk-examples/tree/main/python/example_code/kinesis#code-examples)中查找完整示例，了解如何进行设置和运行。
```
class KinesisStream:
"""Encapsulates a Kinesis stream."""
def __init__(self, kinesis_client):
"""
:param kinesis_client: A Boto3 Kinesis client.
"""
self.kinesis_client = kinesis_client
self.name = None
self.details = None
self.stream_exists_waiter = kinesis_client.get_waiter("stream_exists")
def create(self, name, wait_until_exists=True):
"""
Creates a stream.
:param name: The name of the stream.
:param wait_until_exists: When True, waits until the service reports that
the stream exists, then queries for its metadata.
"""
try:
self.kinesis_client.create_stream(StreamName=name, ShardCount=1)
self.name = name
logger.info("Created stream %s.", name)
if wait_until_exists:
logger.info("Waiting until exists.")
self.stream_exists_waiter.wait(StreamName=name)
self.describe(name)
except ClientError:
logger.exception("Couldn't create stream %s.", name)
raise
```
+ 有关 API 的详细信息，请参阅适用[CreateStream](https://docs.aws.amazon.com/goto/boto3/kinesis-2013-12-02/CreateStream)于 *Python 的AWS SDK (Boto3) API 参考*。
------
#### [ Rust ]
**适用于 Rust 的 SDK**
还有更多相关信息 GitHub。在 [AWS 代码示例存储库](https://github.com/awsdocs/aws-doc-sdk-examples/tree/main/rustv1/examples/kinesis#code-examples)中查找完整示例，了解如何进行设置和运行。
```
async fn make_stream(client: &Client, stream: &str) -> Result<(), Error> {
client
.create_stream()
.stream_name(stream)
.shard_count(4)
.send()
.await?;
println!("Created stream");
Ok(())
}
```
+ 有关 API 的详细信息，请参阅适用[CreateStream](https://docs.rs/aws-sdk-kinesis/latest/aws_sdk_kinesis/client/struct.Client.html#method.create_stream)于 *Rust 的AWS SDK API 参考*。
------
#### [ SAP ABAP ]
**适用于 SAP ABAP 的 SDK**
还有更多相关信息 GitHub。在 [AWS 代码示例存储库](https://github.com/awsdocs/aws-doc-sdk-examples/tree/main/sap-abap/services/kns#code-examples)中查找完整示例，了解如何进行设置和运行。
```
TRY.
lo_kns->createstream(
iv_streamname = iv_stream_name
iv_shardcount = iv_shard_count ).
MESSAGE 'Stream created.' TYPE 'I'.
CATCH /aws1/cx_knsinvalidargumentex.
MESSAGE 'The specified argument was not valid.' TYPE 'E'.
CATCH /aws1/cx_knslimitexceededex.
MESSAGE 'The request processing has failed because of a limit exceed exception.' TYPE 'E'.
CATCH /aws1/cx_knsresourceinuseex.
MESSAGE 'The request processing has failed because the resource is in use.' TYPE 'E'.
ENDTRY.
```
+ 有关 API 的详细信息，请参阅适用[CreateStream](https://docs.aws.amazon.com/sdk-for-sap-abap/v1/api/latest/index.html)于 S *AP 的AWS SDK ABAP API 参考*。
------
# `DeleteStream`与 AWS SDK 或 CLI 配合使用
以下代码示例演示如何使用 `DeleteStream`。
操作示例是大型程序的代码摘录，必须在上下文中运行。在以下代码示例中，您可以查看此操作的上下文：
+ [了解基本功能](kinesis_example_kinesis_Scenario_GettingStarted_section.md)
------
#### [ .NET ]
**适用于 .NET 的 SDK**
还有更多相关信息 GitHub。在 [AWS 代码示例存储库](https://github.com/awsdocs/aws-doc-sdk-examples/tree/main/dotnetv3/Kinesis#code-examples)中查找完整示例，了解如何进行设置和运行。
```
using System;
using System.Threading.Tasks;
using Amazon.Kinesis;
using Amazon.Kinesis.Model;
///
/// Shows how to delete an Amazon Kinesis stream.
///
public class DeleteStream
{
public static async Task Main()
{
IAmazonKinesis client = new AmazonKinesisClient();
string streamName = "AmazonKinesisStream";
var success = await DeleteStreamAsync(client, streamName);
if (success)
{
Console.WriteLine($"Stream, {streamName} successfully deleted.");
}
else
{
Console.WriteLine("Stream not deleted.");
}
}
///
/// Deletes a Kinesis stream.
///
/// An initialized Kinesis client object.
/// The name of the string to delete.
/// A Boolean value representing the success of the operation.
public static async Task DeleteStreamAsync(IAmazonKinesis client, string streamName)
{
// If EnforceConsumerDeletion is true, any consumers
// of this stream will also be deleted. If it is set
// to false and this stream has any consumers, the
// call will fail with a ResourceInUseException.
var request = new DeleteStreamRequest
{
StreamName = streamName,
EnforceConsumerDeletion = true,
};
var response = await client.DeleteStreamAsync(request);
return response.HttpStatusCode == System.Net.HttpStatusCode.OK;
}
}
```
+ 有关 API 的详细信息，请参阅 *适用于 .NET 的 AWS SDK API 参考[DeleteStream](https://docs.aws.amazon.com/goto/DotNetSDKV3/kinesis-2013-12-02/DeleteStream)*中的。
------
#### [ CLI ]
**AWS CLI**
**删除数据流**
以下 `delete-stream` 示例删除指定的数据流。
```
aws kinesis delete-stream \
--stream-name samplestream
```
此命令不生成任何输出。
有关更多信息，请参阅《Amazon Kinesis Data Streams 开发人员指南》**中的[删除流](https://docs.aws.amazon.com/streams/latest/dev/kinesis-using-sdk-java-delete-stream.html)。
+ 有关 API 的详细信息，请参阅*AWS CLI 命令参考[DeleteStream](https://awscli.amazonaws.com/v2/documentation/api/latest/reference/kinesis/delete-stream.html)*中的。
------
#### [ Java ]
**适用于 Java 的 SDK 2.x**
还有更多相关信息 GitHub。在 [AWS 代码示例存储库](https://github.com/awsdocs/aws-doc-sdk-examples/tree/main/javav2/example_code/kinesis#code-examples)中查找完整示例，了解如何进行设置和运行。
```
import software.amazon.awssdk.regions.Region;
import software.amazon.awssdk.services.kinesis.KinesisClient;
import software.amazon.awssdk.services.kinesis.model.DeleteStreamRequest;
import software.amazon.awssdk.services.kinesis.model.KinesisException;
/**
* Before running this Java V2 code example, set up your development
* environment, including your credentials.
*
* For more information, see the following documentation topic:
*
* https://docs.aws.amazon.com/sdk-for-java/latest/developer-guide/get-started.html
*/
public class DeleteDataStream {
public static void main(String[] args) {
final String usage = """
Usage:
Where:
streamName - The Amazon Kinesis data stream (for example, StockTradeStream)
""";
if (args.length != 1) {
System.out.println(usage);
System.exit(1);
}
String streamName = args[0];
Region region = Region.US_EAST_1;
KinesisClient kinesisClient = KinesisClient.builder()
.region(region)
.build();
deleteStream(kinesisClient, streamName);
kinesisClient.close();
System.out.println("Done");
}
public static void deleteStream(KinesisClient kinesisClient, String streamName) {
try {
DeleteStreamRequest delStream = DeleteStreamRequest.builder()
.streamName(streamName)
.build();
kinesisClient.deleteStream(delStream);
} catch (KinesisException e) {
System.err.println(e.getMessage());
System.exit(1);
}
}
}
```
+ 有关 API 的详细信息，请参阅 *AWS SDK for Java 2.x API 参考[DeleteStream](https://docs.aws.amazon.com/goto/SdkForJavaV2/kinesis-2013-12-02/DeleteStream)*中的。
------
#### [ PowerShell ]
**适用于 PowerShell V4 的工具**
**示例 1：删除指定的流。在命令执行之前，系统会提示您进行确认。要取消确认提示，请使用 -Force 开关。**
```
Remove-KINStream -StreamName "mystream"
```
+ 有关 API 的详细信息，请参阅 *AWS Tools for PowerShell Cmdlet 参考 (V* 4) [DeleteStream](https://docs.aws.amazon.com/powershell/v4/reference)中的。
**适用于 PowerShell V5 的工具**
**示例 1：删除指定的流。在命令执行之前，系统会提示您进行确认。要取消确认提示，请使用 -Force 开关。**
```
Remove-KINStream -StreamName "mystream"
```
+ 有关 API 的详细信息，请参阅 *AWS Tools for PowerShell Cmdlet 参考 (V* 5) [DeleteStream](https://docs.aws.amazon.com/powershell/v5/reference)中的。
------
#### [ Python ]
**适用于 Python 的 SDK（Boto3）**
还有更多相关信息 GitHub。在 [AWS 代码示例存储库](https://github.com/awsdocs/aws-doc-sdk-examples/tree/main/python/example_code/kinesis#code-examples)中查找完整示例，了解如何进行设置和运行。
```
class KinesisStream:
"""Encapsulates a Kinesis stream."""
def __init__(self, kinesis_client):
"""
:param kinesis_client: A Boto3 Kinesis client.
"""
self.kinesis_client = kinesis_client
self.name = None
self.details = None
self.stream_exists_waiter = kinesis_client.get_waiter("stream_exists")
def delete(self):
"""
Deletes a stream.
"""
try:
self.kinesis_client.delete_stream(StreamName=self.name)
self._clear()
logger.info("Deleted stream %s.", self.name)
except ClientError:
logger.exception("Couldn't delete stream %s.", self.name)
raise
```
+ 有关 API 的详细信息，请参阅适用[DeleteStream](https://docs.aws.amazon.com/goto/boto3/kinesis-2013-12-02/DeleteStream)于 *Python 的AWS SDK (Boto3) API 参考*。
------
#### [ Rust ]
**适用于 Rust 的 SDK**
还有更多相关信息 GitHub。在 [AWS 代码示例存储库](https://github.com/awsdocs/aws-doc-sdk-examples/tree/main/rustv1/examples/kinesis#code-examples)中查找完整示例，了解如何进行设置和运行。
```
async fn remove_stream(client: &Client, stream: &str) -> Result<(), Error> {
client.delete_stream().stream_name(stream).send().await?;
println!("Deleted stream.");
Ok(())
}
```
+ 有关 API 的详细信息，请参阅适用[DeleteStream](https://docs.rs/aws-sdk-kinesis/latest/aws_sdk_kinesis/client/struct.Client.html#method.delete_stream)于 *Rust 的AWS SDK API 参考*。
------
#### [ SAP ABAP ]
**适用于 SAP ABAP 的 SDK**
还有更多相关信息 GitHub。在 [AWS 代码示例存储库](https://github.com/awsdocs/aws-doc-sdk-examples/tree/main/sap-abap/services/kns#code-examples)中查找完整示例，了解如何进行设置和运行。
```
TRY.
lo_kns->deletestream(
iv_streamname = iv_stream_name ).
MESSAGE 'Stream deleted.' TYPE 'I'.
CATCH /aws1/cx_knslimitexceededex.
MESSAGE 'The request processing has failed because of a limit exceed exception.' TYPE 'E'.
CATCH /aws1/cx_knsresourceinuseex.
MESSAGE 'The request processing has failed because the resource is in use.' TYPE 'E'.
ENDTRY.
```
+ 有关 API 的详细信息，请参阅适用[DeleteStream](https://docs.aws.amazon.com/sdk-for-sap-abap/v1/api/latest/index.html)于 S *AP 的AWS SDK ABAP API 参考*。
------
# `DeregisterStreamConsumer`与 AWS SDK 或 CLI 配合使用
以下代码示例演示如何使用 `DeregisterStreamConsumer`。
------
#### [ .NET ]
**适用于 .NET 的 SDK**
还有更多相关信息 GitHub。在 [AWS 代码示例存储库](https://github.com/awsdocs/aws-doc-sdk-examples/tree/main/dotnetv3/Kinesis#code-examples)中查找完整示例，了解如何进行设置和运行。
```
using System;
using System.Threading.Tasks;
using Amazon.Kinesis;
using Amazon.Kinesis.Model;
///
/// Shows how to deregister a consumer from an Amazon Kinesis stream.
///
public class DeregisterConsumer
{
public static async Task Main(string[] args)
{
IAmazonKinesis client = new AmazonKinesisClient();
string streamARN = "arn:aws:kinesis:us-west-2:000000000000:stream/AmazonKinesisStream";
string consumerName = "CONSUMER_NAME";
string consumerARN = "arn:aws:kinesis:us-west-2:000000000000:stream/AmazonKinesisStream/consumer/CONSUMER_NAME:000000000000";
var success = await DeregisterConsumerAsync(client, streamARN, consumerARN, consumerName);
if (success)
{
Console.WriteLine($"{consumerName} successfully deregistered.");
}
else
{
Console.WriteLine($"{consumerName} was not successfully deregistered.");
}
}
///
/// Deregisters a consumer from a Kinesis stream.
///
/// An initialized Kinesis client object.
/// The ARN of a Kinesis stream.
/// The ARN of the consumer.
/// The name of the consumer.
/// A Boolean value representing the success of the operation.
public static async Task DeregisterConsumerAsync(
IAmazonKinesis client,
string streamARN,
string consumerARN,
string consumerName)
{
var request = new DeregisterStreamConsumerRequest
{
StreamARN = streamARN,
ConsumerARN = consumerARN,
ConsumerName = consumerName,
};
var response = await client.DeregisterStreamConsumerAsync(request);
return response.HttpStatusCode == System.Net.HttpStatusCode.OK;
}
}
```
+ 有关 API 的详细信息，请参阅 *适用于 .NET 的 AWS SDK API 参考[DeregisterStreamConsumer](https://docs.aws.amazon.com/goto/DotNetSDKV3/kinesis-2013-12-02/DeregisterStreamConsumer)*中的。
------
#### [ CLI ]
**AWS CLI**
**注销数据流消费端**
以下 `deregister-stream-consumer` 示例注销指定数据流的指定消费端。
```
aws kinesis deregister-stream-consumer \
--stream-arn arn:aws:kinesis:us-west-2:123456789012:stream/samplestream \
--consumer-name KinesisConsumerApplication
```
此命令不生成任何输出。
有关更多信息，请参阅《Amazon Kinesis Data Streams 开发人员指南》中的[使用 Kinesis Data Streams API 开发具有增强型扇出功能的消费端](https://docs.aws.amazon.com/streams/latest/dev/building-enhanced-consumers-api.html)**。
+ 有关 API 的详细信息，请参阅*AWS CLI 命令参考[DeregisterStreamConsumer](https://awscli.amazonaws.com/v2/documentation/api/latest/reference/kinesis/deregister-stream-consumer.html)*中的。
------
# `DescribeStream`与 AWS SDK 或 CLI 配合使用
以下代码示例演示如何使用 `DescribeStream`。
------
#### [ CLI ]
**AWS CLI**
**描述数据流**
以下 `describe-stream` 示例返回指定数据流的详细信息。
```
aws kinesis describe-stream \
--stream-name samplestream
```
输出：
```
{
"StreamDescription": {
"Shards": [
{
"ShardId": "shardId-000000000000",
"HashKeyRange": {
"StartingHashKey": "0",
"EndingHashKey": "113427455640312821154458202477256070484"
},
"SequenceNumberRange": {
"StartingSequenceNumber": "49600871682957036442365024926191073437251060580128653314"
}
},
{
"ShardId": "shardId-000000000001",
"HashKeyRange": {
"StartingHashKey": "113427455640312821154458202477256070485",
"EndingHashKey": "226854911280625642308916404954512140969"
},
"SequenceNumberRange": {
"StartingSequenceNumber": "49600871682979337187563555549332609155523708941634633746"
}
},
{
"ShardId": "shardId-000000000002",
"HashKeyRange": {
"StartingHashKey": "226854911280625642308916404954512140970",
"EndingHashKey": "340282366920938463463374607431768211455"
},
"SequenceNumberRange": {
"StartingSequenceNumber": "49600871683001637932762086172474144873796357303140614178"
}
}
],
"StreamARN": "arn:aws:kinesis:us-west-2:123456789012:stream/samplestream",
"StreamName": "samplestream",
"StreamStatus": "ACTIVE",
"RetentionPeriodHours": 24,
"EnhancedMonitoring": [
{
"ShardLevelMetrics": []
}
],
"EncryptionType": "NONE",
"KeyId": null,
"StreamCreationTimestamp": 1572297168.0
}
}
```
有关更多信息，请参阅《Amazon Kinesis Data Streams 开发人员指南》**中的[创建和管理流](https://docs.aws.amazon.com/streams/latest/dev/working-with-streams.html)。
+ 有关 API 的详细信息，请参阅*AWS CLI 命令参考[DescribeStream](https://awscli.amazonaws.com/v2/documentation/api/latest/reference/kinesis/describe-stream.html)*中的。
------
#### [ PowerShell ]
**适用于 PowerShell V4 的工具**
**示例 1：返回指定流的详细信息。**
```
Get-KINStream -StreamName "mystream"
```
**输出**：
```
HasMoreShards : False
RetentionPeriodHours : 24
Shards : {}
StreamARN : arn:aws:kinesis:us-west-2:123456789012:stream/mystream
StreamName : mystream
StreamStatus : ACTIVE
```
+ 有关 API 的详细信息，请参阅 *AWS Tools for PowerShell Cmdlet 参考 (V* 4) [DescribeStream](https://docs.aws.amazon.com/powershell/v4/reference)中的。
**适用于 PowerShell V5 的工具**
**示例 1：返回指定流的详细信息。**
```
Get-KINStream -StreamName "mystream"
```
**输出**：
```
HasMoreShards : False
RetentionPeriodHours : 24
Shards : {}
StreamARN : arn:aws:kinesis:us-west-2:123456789012:stream/mystream
StreamName : mystream
StreamStatus : ACTIVE
```
+ 有关 API 的详细信息，请参阅 *AWS Tools for PowerShell Cmdlet 参考 (V* 5) [DescribeStream](https://docs.aws.amazon.com/powershell/v5/reference)中的。
------
#### [ Python ]
**适用于 Python 的 SDK（Boto3）**
还有更多相关信息 GitHub。在 [AWS 代码示例存储库](https://github.com/awsdocs/aws-doc-sdk-examples/tree/main/python/example_code/kinesis#code-examples)中查找完整示例，了解如何进行设置和运行。
```
class KinesisStream:
"""Encapsulates a Kinesis stream."""
def __init__(self, kinesis_client):
"""
:param kinesis_client: A Boto3 Kinesis client.
"""
self.kinesis_client = kinesis_client
self.name = None
self.details = None
self.stream_exists_waiter = kinesis_client.get_waiter("stream_exists")
def describe(self, name):
"""
Gets metadata about a stream.
:param name: The name of the stream.
:return: Metadata about the stream.
"""
try:
response = self.kinesis_client.describe_stream(StreamName=name)
self.name = name
self.details = response["StreamDescription"]
logger.info("Got stream %s.", name)
except ClientError:
logger.exception("Couldn't get %s.", name)
raise
else:
return self.details
```
+ 有关 API 的详细信息，请参阅适用[DescribeStream](https://docs.aws.amazon.com/goto/boto3/kinesis-2013-12-02/DescribeStream)于 *Python 的AWS SDK (Boto3) API 参考*。
------
#### [ Rust ]
**适用于 Rust 的 SDK**
还有更多相关信息 GitHub。在 [AWS 代码示例存储库](https://github.com/awsdocs/aws-doc-sdk-examples/tree/main/rustv1/examples/kinesis#code-examples)中查找完整示例，了解如何进行设置和运行。
```
async fn show_stream(client: &Client, stream: &str) -> Result<(), Error> {
let resp = client.describe_stream().stream_name(stream).send().await?;
let desc = resp.stream_description.unwrap();
println!("Stream description:");
println!(" Name: {}:", desc.stream_name());
println!(" Status: {:?}", desc.stream_status());
println!(" Open shards: {:?}", desc.shards.len());
println!(" Retention (hours): {}", desc.retention_period_hours());
println!(" Encryption: {:?}", desc.encryption_type.unwrap());
Ok(())
}
```
+ 有关 API 的详细信息，请参阅适用[DescribeStream](https://docs.rs/aws-sdk-kinesis/latest/aws_sdk_kinesis/client/struct.Client.html#method.describe_stream)于 *Rust 的AWS SDK API 参考*。
------
#### [ SAP ABAP ]
**适用于 SAP ABAP 的 SDK**
还有更多相关信息 GitHub。在 [AWS 代码示例存储库](https://github.com/awsdocs/aws-doc-sdk-examples/tree/main/sap-abap/services/kns#code-examples)中查找完整示例，了解如何进行设置和运行。
```
TRY.
oo_result = lo_kns->describestream(
iv_streamname = iv_stream_name ).
DATA(lt_stream_description) = oo_result->get_streamdescription( ).
MESSAGE 'Streams retrieved.' TYPE 'I'.
CATCH /aws1/cx_knslimitexceededex.
MESSAGE 'The request processing has failed because of a limit exceed exception.' TYPE 'E'.
CATCH /aws1/cx_knsresourcenotfoundex.
MESSAGE 'Resource being accessed is not found.' TYPE 'E'.
ENDTRY.
```
+ 有关 API 的详细信息，请参阅适用[DescribeStream](https://docs.aws.amazon.com/sdk-for-sap-abap/v1/api/latest/index.html)于 S *AP 的AWS SDK ABAP API 参考*。
------
# `GetRecords`与 AWS SDK 或 CLI 配合使用
以下代码示例演示如何使用 `GetRecords`。
操作示例是大型程序的代码摘录，必须在上下文中运行。在以下代码示例中，您可以查看此操作的上下文：
+ [了解基本功能](kinesis_example_kinesis_Scenario_GettingStarted_section.md)
------
#### [ CLI ]
**AWS CLI**
**获取分片中的记录**
以下 `get-records` 示例使用指定分片迭代器从 Kinesis 数据流的分片中获取数据记录。
```
aws kinesis get-records \
--shard-iterator AAAAAAAAAAF7/0mWD7IuHj1yGv/TKuNgx2ukD5xipCY4cy4gU96orWwZwcSXh3K9tAmGYeOZyLZrvzzeOFVf9iN99hUPw/w/b0YWYeehfNvnf1DYt5XpDJghLKr3DzgznkTmMymDP3R+3wRKeuEw6/kdxY2yKJH0veaiekaVc4N2VwK/GvaGP2Hh9Fg7N++q0Adg6fIDQPt4p8RpavDbk+A4sL9SWGE1
```
输出：
```
{
"Records": [],
"MillisBehindLatest": 80742000
}
```
有关更多信息，请参阅*亚马逊 Kinesis Data Streams 开发者指南中的使用 Kinesis Data Streams* [API 和 Java 开发工具包开发消费者](https://docs.aws.amazon.com/streams/latest/dev/developing-consumers-with-sdk.html)。 AWS
+ 有关 API 的详细信息，请参阅*AWS CLI 命令参考[GetRecords](https://awscli.amazonaws.com/v2/documentation/api/latest/reference/kinesis/get-records.html)*中的。
------
#### [ Java ]
**适用于 Java 的 SDK 2.x**
还有更多相关信息 GitHub。在 [AWS 代码示例存储库](https://github.com/awsdocs/aws-doc-sdk-examples/tree/main/javav2/example_code/kinesis#code-examples)中查找完整示例，了解如何进行设置和运行。
```
import software.amazon.awssdk.core.SdkBytes;
import software.amazon.awssdk.regions.Region;
import software.amazon.awssdk.services.kinesis.KinesisClient;
import software.amazon.awssdk.services.kinesis.model.DescribeStreamResponse;
import software.amazon.awssdk.services.kinesis.model.DescribeStreamRequest;
import software.amazon.awssdk.services.kinesis.model.Shard;
import software.amazon.awssdk.services.kinesis.model.GetShardIteratorRequest;
import software.amazon.awssdk.services.kinesis.model.GetShardIteratorResponse;
import software.amazon.awssdk.services.kinesis.model.Record;
import software.amazon.awssdk.services.kinesis.model.GetRecordsRequest;
import software.amazon.awssdk.services.kinesis.model.GetRecordsResponse;
import java.util.ArrayList;
import java.util.List;
/**
* Before running this Java V2 code example, set up your development
* environment, including your credentials.
*
* For more information, see the following documentation topic:
*
* https://docs.aws.amazon.com/sdk-for-java/latest/developer-guide/get-started.html
*/
public class GetRecords {
public static void main(String[] args) {
final String usage = """
Usage:
Where:
streamName - The Amazon Kinesis data stream to read from (for example, StockTradeStream).
""";
if (args.length != 1) {
System.out.println(usage);
System.exit(1);
}
String streamName = args[0];
Region region = Region.US_EAST_1;
KinesisClient kinesisClient = KinesisClient.builder()
.region(region)
.build();
getStockTrades(kinesisClient, streamName);
kinesisClient.close();
}
public static void getStockTrades(KinesisClient kinesisClient, String streamName) {
String shardIterator;
String lastShardId = null;
DescribeStreamRequest describeStreamRequest = DescribeStreamRequest.builder()
.streamName(streamName)
.build();
List shards = new ArrayList<>();
DescribeStreamResponse streamRes;
do {
streamRes = kinesisClient.describeStream(describeStreamRequest);
shards.addAll(streamRes.streamDescription().shards());
if (shards.size() > 0) {
lastShardId = shards.get(shards.size() - 1).shardId();
}
} while (streamRes.streamDescription().hasMoreShards());
GetShardIteratorRequest itReq = GetShardIteratorRequest.builder()
.streamName(streamName)
.shardIteratorType("TRIM_HORIZON")
.shardId(lastShardId)
.build();
GetShardIteratorResponse shardIteratorResult = kinesisClient.getShardIterator(itReq);
shardIterator = shardIteratorResult.shardIterator();
// Continuously read data records from shard.
List records;
// Create new GetRecordsRequest with existing shardIterator.
// Set maximum records to return to 1000.
GetRecordsRequest recordsRequest = GetRecordsRequest.builder()
.shardIterator(shardIterator)
.limit(1000)
.build();
GetRecordsResponse result = kinesisClient.getRecords(recordsRequest);
// Put result into record list. Result may be empty.
records = result.records();
// Print records
for (Record record : records) {
SdkBytes byteBuffer = record.data();
System.out.printf("Seq No: %s - %s%n", record.sequenceNumber(), new String(byteBuffer.asByteArray()));
}
}
}
```
+ 有关 API 的详细信息，请参阅 *AWS SDK for Java 2.x API 参考[GetRecords](https://docs.aws.amazon.com/goto/SdkForJavaV2/kinesis-2013-12-02/GetRecords)*中的。
------
#### [ PowerShell ]
**适用于 PowerShell V4 的工具**
**示例 1：此示例将展示如何从一条或多条记录中返回和提取数据。提供的迭代器 Get-KINRecord 用于确定要返回的记录的起始位置，在本例中，这些记录被捕获到变量 \$1records 中。然后，可以通过对 \$1records 集合进行索引来访问每条记录。假设记录中的数据是 UTF-8 编码的文本，则最后一个命令显示如何从对象 MemoryStream 中提取数据并将其作为文本返回到控制台。**
```
$records
$records = Get-KINRecord -ShardIterator "AAAAAAAAAAGIc....9VnbiRNaP"
```
**输出**：
```
MillisBehindLatest NextShardIterator Records
------------------ ----------------- -------
0 AAAAAAAAAAERNIq...uDn11HuUs {Key1, Key2}
```
```
$records.Records[0]
```
**输出**：
```
ApproximateArrivalTimestamp Data PartitionKey SequenceNumber
--------------------------- ---- ------------ --------------
3/7/2016 5:14:33 PM System.IO.MemoryStream Key1 4955986459776...931586
```
```
[Text.Encoding]::UTF8.GetString($records.Records[0].Data.ToArray())
```
**输出**：
```
test data from string
```
+ 有关 API 的详细信息，请参阅 *AWS Tools for PowerShell Cmdlet 参考 (V* 4) [GetRecords](https://docs.aws.amazon.com/powershell/v4/reference)中的。
**适用于 PowerShell V5 的工具**
**示例 1：此示例将展示如何从一条或多条记录中返回和提取数据。提供的迭代器 Get-KINRecord 用于确定要返回的记录的起始位置，在本例中，这些记录被捕获到变量 \$1records 中。然后，可以通过对 \$1records 集合进行索引来访问每条记录。假设记录中的数据是 UTF-8 编码的文本，则最后一个命令显示如何从对象 MemoryStream 中提取数据并将其作为文本返回到控制台。**
```
$records
$records = Get-KINRecord -ShardIterator "AAAAAAAAAAGIc....9VnbiRNaP"
```
**输出**：
```
MillisBehindLatest NextShardIterator Records
------------------ ----------------- -------
0 AAAAAAAAAAERNIq...uDn11HuUs {Key1, Key2}
```
```
$records.Records[0]
```
**输出**：
```
ApproximateArrivalTimestamp Data PartitionKey SequenceNumber
--------------------------- ---- ------------ --------------
3/7/2016 5:14:33 PM System.IO.MemoryStream Key1 4955986459776...931586
```
```
[Text.Encoding]::UTF8.GetString($records.Records[0].Data.ToArray())
```
**输出**：
```
test data from string
```
+ 有关 API 的详细信息，请参阅 *AWS Tools for PowerShell Cmdlet 参考 (V* 5) [GetRecords](https://docs.aws.amazon.com/powershell/v5/reference)中的。
------
#### [ Python ]
**适用于 Python 的 SDK（Boto3）**
还有更多相关信息 GitHub。在 [AWS 代码示例存储库](https://github.com/awsdocs/aws-doc-sdk-examples/tree/main/python/example_code/kinesis#code-examples)中查找完整示例，了解如何进行设置和运行。
```
class KinesisStream:
"""Encapsulates a Kinesis stream."""
def __init__(self, kinesis_client):
"""
:param kinesis_client: A Boto3 Kinesis client.
"""
self.kinesis_client = kinesis_client
self.name = None
self.details = None
self.stream_exists_waiter = kinesis_client.get_waiter("stream_exists")
def get_records(self, max_records):
"""
Gets records from the stream. This function is a generator that first gets
a shard iterator for the stream, then uses the shard iterator to get records
in batches from the stream. The shard iterator can be accessed through the
'details' property, which is populated using the 'describe' function of this class.
Each batch of records is yielded back to the caller until the specified
maximum number of records has been retrieved.
:param max_records: The maximum number of records to retrieve.
:return: Yields the current batch of retrieved records.
"""
try:
response = self.kinesis_client.get_shard_iterator(
StreamName=self.name,
ShardId=self.details["Shards"][0]["ShardId"],
ShardIteratorType="LATEST",
)
shard_iter = response["ShardIterator"]
record_count = 0
while record_count < max_records:
response = self.kinesis_client.get_records(
ShardIterator=shard_iter, Limit=10
)
shard_iter = response["NextShardIterator"]
records = response["Records"]
logger.info("Got %s records.", len(records))
record_count += len(records)
yield records
except ClientError:
logger.exception("Couldn't get records from stream %s.", self.name)
raise
def describe(self, name):
"""
Gets metadata about a stream.
:param name: The name of the stream.
:return: Metadata about the stream.
"""
try:
response = self.kinesis_client.describe_stream(StreamName=name)
self.name = name
self.details = response["StreamDescription"]
logger.info("Got stream %s.", name)
except ClientError:
logger.exception("Couldn't get %s.", name)
raise
else:
return self.details
```
+ 有关 API 的详细信息，请参阅适用[GetRecords](https://docs.aws.amazon.com/goto/boto3/kinesis-2013-12-02/GetRecords)于 *Python 的AWS SDK (Boto3) API 参考*。
------
#### [ SAP ABAP ]
**适用于 SAP ABAP 的 SDK**
还有更多相关信息 GitHub。在 [AWS 代码示例存储库](https://github.com/awsdocs/aws-doc-sdk-examples/tree/main/sap-abap/services/kns#code-examples)中查找完整示例，了解如何进行设置和运行。
```
TRY.
oo_result = lo_kns->getrecords( " oo_result is returned for testing purposes. "
iv_sharditerator = iv_shard_iterator ).
DATA(lt_records) = oo_result->get_records( ).
MESSAGE 'Record retrieved.' TYPE 'I'.
CATCH /aws1/cx_knsexpirediteratorex.
MESSAGE 'Iterator expired.' TYPE 'E'.
CATCH /aws1/cx_knsinvalidargumentex.
MESSAGE 'The specified argument was not valid.' TYPE 'E'.
CATCH /aws1/cx_knskmsaccessdeniedex.
MESSAGE 'You do not have permission to perform this AWS KMS action.' TYPE 'E'.
CATCH /aws1/cx_knskmsdisabledex.
MESSAGE 'KMS key used is disabled.' TYPE 'E'.
CATCH /aws1/cx_knskmsinvalidstateex.
MESSAGE 'KMS key used is in an invalid state. ' TYPE 'E'.
CATCH /aws1/cx_knskmsnotfoundex.
MESSAGE 'KMS key used is not found.' TYPE 'E'.
CATCH /aws1/cx_knskmsoptinrequired.
MESSAGE 'KMS key option is required.' TYPE 'E'.
CATCH /aws1/cx_knskmsthrottlingex.
MESSAGE 'The rate of requests to AWS KMS is exceeding the request quotas.' TYPE 'E'.
CATCH /aws1/cx_knsprovthruputexcdex.
MESSAGE 'The request rate for the stream is too high, or the requested data is too large for the available throughput.' TYPE 'E'.
CATCH /aws1/cx_knsresourcenotfoundex.
MESSAGE 'Resource being accessed is not found.' TYPE 'E'.
ENDTRY.
```
+ 有关 API 的详细信息，请参阅适用[GetRecords](https://docs.aws.amazon.com/sdk-for-sap-abap/v1/api/latest/index.html)于 S *AP 的AWS SDK ABAP API 参考*。
------
# 将 `GetShardIterator` 与 CLI 配合使用
以下代码示例演示如何使用 `GetShardIterator`。
操作示例是大型程序的代码摘录，必须在上下文中运行。在以下代码示例中，您可以查看此操作的上下文：
+ [了解基本功能](kinesis_example_kinesis_Scenario_GettingStarted_section.md)
------
#### [ CLI ]
**AWS CLI**
**获取分片迭代器**
以下 `get-shard-iterator` 示例使用 `AT_SEQUENCE_NUMBER` 分片迭代器类型并生成分片迭代器，以便从指定序列号所表示的位置开始精确读取数据记录。
```
aws kinesis get-shard-iterator \
--stream-name samplestream \
--shard-id shardId-000000000001 \
--shard-iterator-type LATEST
```
输出：
```
{
"ShardIterator": "AAAAAAAAAAFEvJjIYI+3jw/4aqgH9FifJ+n48XWTh/IFIsbILP6o5eDueD39NXNBfpZ10WL5K6ADXk8w+5H+Qhd9cFA9k268CPXCz/kebq1TGYI7Vy+lUkA9BuN3xvATxMBGxRY3zYK05gqgvaIRn94O8SqeEqwhigwZxNWxID3Ej7YYYcxQi8Q/fIrCjGAy/n2r5Z9G864YpWDfN9upNNQAR/iiOWKs"
}
```
有关更多信息，请参阅*亚马逊 Kinesis Data Streams 开发者指南中的使用 Kinesis Data Streams* [API 和 Java 开发工具包开发消费者](https://docs.aws.amazon.com/streams/latest/dev/developing-consumers-with-sdk.html)。 AWS
+ 有关 API 的详细信息，请参阅*AWS CLI 命令参考[GetShardIterator](https://awscli.amazonaws.com/v2/documentation/api/latest/reference/kinesis/get-shard-iterator.html)*中的。
------
#### [ PowerShell ]
**适用于 PowerShell V4 的工具**
**示例 1：返回指定分片和起始位置的分片迭代器。通过引用返回的流对象的 Shards 集合，可以从 Get-KINStream cmdlet 的输出中获取分片标识符和序列号的详细信息。返回的迭代器可以与 Get-KINRecord cmdlet 一起使用，在分片中提取数据记录。**
```
Get-KINShardIterator -StreamName "mystream" -ShardId "shardId-000000000000" -ShardIteratorType AT_SEQUENCE_NUMBER -StartingSequenceNumber "495598645..."
```
**输出**：
```
AAAAAAAAAAGIc....9VnbiRNaP
```
+ 有关 API 的详细信息，请参阅 *AWS Tools for PowerShell Cmdlet 参考 (V* 4) [GetShardIterator](https://docs.aws.amazon.com/powershell/v4/reference)中的。
**适用于 PowerShell V5 的工具**
**示例 1：返回指定分片和起始位置的分片迭代器。通过引用返回的流对象的 Shards 集合，可以从 Get-KINStream cmdlet 的输出中获取分片标识符和序列号的详细信息。返回的迭代器可以与 Get-KINRecord cmdlet 一起使用，在分片中提取数据记录。**
```
Get-KINShardIterator -StreamName "mystream" -ShardId "shardId-000000000000" -ShardIteratorType AT_SEQUENCE_NUMBER -StartingSequenceNumber "495598645..."
```
**输出**：
```
AAAAAAAAAAGIc....9VnbiRNaP
```
+ 有关 API 的详细信息，请参阅 *AWS Tools for PowerShell Cmdlet 参考 (V* 5) [GetShardIterator](https://docs.aws.amazon.com/powershell/v5/reference)中的。
------
# 与 AWS SDK `ListStreamConsumers` 配合使用
以下代码示例演示了如何使用 `ListStreamConsumers`。
------
#### [ .NET ]
**适用于 .NET 的 SDK**
还有更多相关信息 GitHub。在 [AWS 代码示例存储库](https://github.com/awsdocs/aws-doc-sdk-examples/tree/main/dotnetv3/Kinesis#code-examples)中查找完整示例，了解如何进行设置和运行。
```
using System;
using System.Collections.Generic;
using System.Threading.Tasks;
using Amazon.Kinesis;
using Amazon.Kinesis.Model;
///
/// List the consumers of an Amazon Kinesis stream.
///
public class ListConsumers
{
public static async Task Main()
{
IAmazonKinesis client = new AmazonKinesisClient();
string streamARN = "arn:aws:kinesis:us-east-2:000000000000:stream/AmazonKinesisStream";
int maxResults = 10;
var consumers = await ListConsumersAsync(client, streamARN, maxResults);
if (consumers.Count > 0)
{
consumers
.ForEach(c => Console.WriteLine($"Name: {c.ConsumerName} ARN: {c.ConsumerARN}"));
}
else
{
Console.WriteLine("No consumers found.");
}
}
///
/// Retrieve a list of the consumers for a Kinesis stream.
///
/// An initialized Kinesis client object.
/// The ARN of the stream for which we want to
/// retrieve a list of clients.
/// The maximum number of results to return.
/// A list of Consumer objects.
public static async Task> ListConsumersAsync(IAmazonKinesis client, string streamARN, int maxResults)
{
var request = new ListStreamConsumersRequest
{
StreamARN = streamARN,
MaxResults = maxResults,
};
var response = await client.ListStreamConsumersAsync(request);
return response.Consumers;
}
}
```
+ 有关 API 的详细信息，请参阅 *适用于 .NET 的 AWS SDK API 参考[ListStreamConsumers](https://docs.aws.amazon.com/goto/DotNetSDKV3/kinesis-2013-12-02/ListStreamConsumers)*中的。
------
# `ListStreams`与 AWS SDK 或 CLI 配合使用
以下代码示例演示如何使用 `ListStreams`。
------
#### [ .NET ]
**适用于 .NET 的 SDK**
还有更多相关信息 GitHub。在 [AWS 代码示例存储库](https://github.com/awsdocs/aws-doc-sdk-examples/tree/main/dotnetv3/Kinesis#code-examples)中查找完整示例，了解如何进行设置和运行。
```
using System;
using System.Collections.Generic;
using System.Threading.Tasks;
using Amazon.Kinesis;
using Amazon.Kinesis.Model;
///
/// Retrieves and displays a list of existing Amazon Kinesis streams.
///
public class ListStreams
{
public static async Task Main(string[] args)
{
IAmazonKinesis client = new AmazonKinesisClient();
var response = await client.ListStreamsAsync(new ListStreamsRequest());
List streamNames = response.StreamNames;
if (streamNames.Count > 0)
{
streamNames
.ForEach(s => Console.WriteLine($"Stream name: {s}"));
}
else
{
Console.WriteLine("No streams were found.");
}
}
}
```
+ 有关 API 的详细信息，请参阅 *适用于 .NET 的 AWS SDK API 参考[ListStreams](https://docs.aws.amazon.com/goto/DotNetSDKV3/kinesis-2013-12-02/ListStreams)*中的。
------
#### [ CLI ]
**AWS CLI**
**列出数据流**
以下 `list-streams` 示例列出了当前账户和区域中的所有活动数据流。
```
aws kinesis list-streams
```
输出：
```
{
"StreamNames": [
"samplestream",
"samplestream1"
]
}
```
有关更多信息，请参阅《Amazon Kinesis Data Streams 开发人员指南》**中的[列出流](https://docs.aws.amazon.com/streams/latest/dev/kinesis-using-sdk-java-list-streams.html)。
+ 有关 API 的详细信息，请参阅*AWS CLI 命令参考[ListStreams](https://awscli.amazonaws.com/v2/documentation/api/latest/reference/kinesis/list-streams.html)*中的。
------
#### [ Rust ]
**适用于 Rust 的 SDK**
还有更多相关信息 GitHub。在 [AWS 代码示例存储库](https://github.com/awsdocs/aws-doc-sdk-examples/tree/main/rustv1/examples/kinesis#code-examples)中查找完整示例，了解如何进行设置和运行。
```
async fn show_streams(client: &Client) -> Result<(), Error> {
let resp = client.list_streams().send().await?;
println!("Stream names:");
let streams = resp.stream_names;
for stream in &streams {
println!(" {}", stream);
}
println!("Found {} stream(s)", streams.len());
Ok(())
}
```
+ 有关 API 的详细信息，请参阅适用[ListStreams](https://docs.rs/aws-sdk-kinesis/latest/aws_sdk_kinesis/client/struct.Client.html#method.list_streams)于 *Rust 的AWS SDK API 参考*。
------
#### [ SAP ABAP ]
**适用于 SAP ABAP 的 SDK**
还有更多相关信息 GitHub。在 [AWS 代码示例存储库](https://github.com/awsdocs/aws-doc-sdk-examples/tree/main/sap-abap/services/kns#code-examples)中查找完整示例，了解如何进行设置和运行。
```
TRY.
oo_result = lo_kns->liststreams( " oo_result is returned for testing purposes. "
"Set Limit to specify that a maximum of streams should be returned."
iv_limit = iv_limit ).
DATA(lt_streams) = oo_result->get_streamnames( ).
MESSAGE 'Streams listed.' TYPE 'I'.
CATCH /aws1/cx_knslimitexceededex.
MESSAGE 'The request processing has failed because of a limit exceed exception.' TYPE 'E'.
ENDTRY.
```
+ 有关 API 的详细信息，请参阅适用[ListStreams](https://docs.aws.amazon.com/sdk-for-sap-abap/v1/api/latest/index.html)于 S *AP 的AWS SDK ABAP API 参考*。
------
# `ListTagsForStream`与 AWS SDK 或 CLI 配合使用
以下代码示例演示如何使用 `ListTagsForStream`。
------
#### [ .NET ]
**适用于 .NET 的 SDK**
还有更多相关信息 GitHub。在 [AWS 代码示例存储库](https://github.com/awsdocs/aws-doc-sdk-examples/tree/main/dotnetv3/Kinesis#code-examples)中查找完整示例，了解如何进行设置和运行。
```
using System;
using System.Collections.Generic;
using System.Threading.Tasks;
using Amazon.Kinesis;
using Amazon.Kinesis.Model;
///
/// Shows how to list the tags that have been attached to an Amazon Kinesis
/// stream.
///
public class ListTags
{
public static async Task Main()
{
IAmazonKinesis client = new AmazonKinesisClient();
string streamName = "AmazonKinesisStream";
await ListTagsAsync(client, streamName);
}
///
/// List the tags attached to a Kinesis stream.
///
/// An initialized Kinesis client object.
/// The name of the Kinesis stream for which you
/// wish to display tags.
public static async Task ListTagsAsync(IAmazonKinesis client, string streamName)
{
var request = new ListTagsForStreamRequest
{
StreamName = streamName,
Limit = 10,
};
var response = await client.ListTagsForStreamAsync(request);
DisplayTags(response.Tags);
while (response.HasMoreTags)
{
request.ExclusiveStartTagKey = response.Tags[response.Tags.Count - 1].Key;
response = await client.ListTagsForStreamAsync(request);
}
}
///
/// Displays the items in a list of Kinesis tags.
///
/// A list of the Tag objects to be displayed.
public static void DisplayTags(List tags)
{
tags
.ForEach(t => Console.WriteLine($"Key: {t.Key} Value: {t.Value}"));
}
}
```
+ 有关 API 的详细信息，请参阅 *适用于 .NET 的 AWS SDK API 参考[ListTagsForStream](https://docs.aws.amazon.com/goto/DotNetSDKV3/kinesis-2013-12-02/ListTagsForStream)*中的。
------
#### [ CLI ]
**AWS CLI**
**列出数据流的标签**
以下 `list-tags-for-stream` 示例列出附加到指定数据流的标签：
```
aws kinesis list-tags-for-stream \
--stream-name samplestream
```
输出：
```
{
"Tags": [
{
"Key": "samplekey",
"Value": "example"
}
],
"HasMoreTags": false
}
```
有关更多信息，请参阅《Amazon Kinesis Data Streams 开发人员指南》**中的[标记流](https://docs.aws.amazon.com/streams/latest/dev/tagging.html)。
+ 有关 API 的详细信息，请参阅*AWS CLI 命令参考[ListTagsForStream](https://awscli.amazonaws.com/v2/documentation/api/latest/reference/kinesis/list-tags-for-stream.html)*中的。
------
# `PutRecord`与 AWS SDK 或 CLI 配合使用
以下代码示例演示如何使用 `PutRecord`。
操作示例是大型程序的代码摘录，必须在上下文中运行。在以下代码示例中，您可以查看此操作的上下文：
+ [了解基本功能](kinesis_example_kinesis_Scenario_GettingStarted_section.md)
------
#### [ CLI ]
**AWS CLI**
**将记录写入数据流**
以下 `put-record` 示例使用指定的分区键将单个数据记录写入指定的数据流。
```
aws kinesis put-record \
--stream-name samplestream \
--data sampledatarecord \
--partition-key samplepartitionkey
```
输出：
```
{
"ShardId": "shardId-000000000009",
"SequenceNumber": "49600902273357540915989931256901506243878407835297513618",
"EncryptionType": "KMS"
}
```
有关更多信息，请参阅[《亚马逊 Kinesis Data Streams 开发者指南》中的使用亚马逊 Kinesis Data Streams API 和 Java 开发工具包开发生产](https://docs.aws.amazon.com/streams/latest/dev/developing-producers-with-sdk.html)*者*。 AWS
+ 有关 API 的详细信息，请参阅*AWS CLI 命令参考[PutRecord](https://awscli.amazonaws.com/v2/documentation/api/latest/reference/kinesis/put-record.html)*中的。
------
#### [ Java ]
**适用于 Java 的 SDK 2.x**
还有更多相关信息 GitHub。在 [AWS 代码示例存储库](https://github.com/awsdocs/aws-doc-sdk-examples/tree/main/javav2/example_code/kinesis#code-examples)中查找完整示例，了解如何进行设置和运行。
```
import software.amazon.awssdk.core.SdkBytes;
import software.amazon.awssdk.regions.Region;
import software.amazon.awssdk.services.kinesis.KinesisClient;
import software.amazon.awssdk.services.kinesis.model.PutRecordRequest;
import software.amazon.awssdk.services.kinesis.model.KinesisException;
import software.amazon.awssdk.services.kinesis.model.DescribeStreamRequest;
import software.amazon.awssdk.services.kinesis.model.DescribeStreamResponse;
/**
* Before running this Java V2 code example, set up your development
* environment, including your credentials.
*
* For more information, see the following documentation topic:
*
* https://docs.aws.amazon.com/sdk-for-java/latest/developer-guide/get-started.html
*/
public class StockTradesWriter {
public static void main(String[] args) {
final String usage = """
Usage:
Where:
streamName - The Amazon Kinesis data stream to which records are written (for example, StockTradeStream)
""";
if (args.length != 1) {
System.out.println(usage);
System.exit(1);
}
String streamName = args[0];
Region region = Region.US_EAST_1;
KinesisClient kinesisClient = KinesisClient.builder()
.region(region)
.build();
// Ensure that the Kinesis Stream is valid.
validateStream(kinesisClient, streamName);
setStockData(kinesisClient, streamName);
kinesisClient.close();
}
public static void setStockData(KinesisClient kinesisClient, String streamName) {
try {
// Repeatedly send stock trades with a 100 milliseconds wait in between.
StockTradeGenerator stockTradeGenerator = new StockTradeGenerator();
// Put in 50 Records for this example.
int index = 50;
for (int x = 0; x < index; x++) {
StockTrade trade = stockTradeGenerator.getRandomTrade();
sendStockTrade(trade, kinesisClient, streamName);
Thread.sleep(100);
}
} catch (KinesisException | InterruptedException e) {
System.err.println(e.getMessage());
System.exit(1);
}
System.out.println("Done");
}
private static void sendStockTrade(StockTrade trade, KinesisClient kinesisClient,
String streamName) {
byte[] bytes = trade.toJsonAsBytes();
// The bytes could be null if there is an issue with the JSON serialization by
// the Jackson JSON library.
if (bytes == null) {
System.out.println("Could not get JSON bytes for stock trade");
return;
}
System.out.println("Putting trade: " + trade);
PutRecordRequest request = PutRecordRequest.builder()
.partitionKey(trade.getTickerSymbol()) // We use the ticker symbol as the partition key, explained in
// the Supplemental Information section below.
.streamName(streamName)
.data(SdkBytes.fromByteArray(bytes))
.build();
try {
kinesisClient.putRecord(request);
} catch (KinesisException e) {
System.err.println(e.getMessage());
}
}
private static void validateStream(KinesisClient kinesisClient, String streamName) {
try {
DescribeStreamRequest describeStreamRequest = DescribeStreamRequest.builder()
.streamName(streamName)
.build();
DescribeStreamResponse describeStreamResponse = kinesisClient.describeStream(describeStreamRequest);
if (!describeStreamResponse.streamDescription().streamStatus().toString().equals("ACTIVE")) {
System.err.println("Stream " + streamName + " is not active. Please wait a few moments and try again.");
System.exit(1);
}
} catch (KinesisException e) {
System.err.println("Error found while describing the stream " + streamName);
System.err.println(e);
System.exit(1);
}
}
}
```
+ 有关 API 的详细信息，请参阅 *AWS SDK for Java 2.x API 参考[PutRecord](https://docs.aws.amazon.com/goto/SdkForJavaV2/kinesis-2013-12-02/PutRecord)*中的。
------
#### [ PowerShell ]
**适用于 PowerShell V4 的工具**
**示例 1：写入一条包含提供给 -Text 参数的字符串的记录。**
```
Write-KINRecord -Text "test data from string" -StreamName "mystream" -PartitionKey "Key1"
```
**示例 2：写入包含指定文件所包含数据的记录。该文件被视为字节序列，因此，如果它包含文本，则在将其与此 cmdlet 搭配使用之前，应使用任何必要的编码进行编写。**
```
Write-KINRecord -FilePath "C:\TestData.txt" -StreamName "mystream" -PartitionKey "Key2"
```
+ 有关 API 的详细信息，请参阅 *AWS Tools for PowerShell Cmdlet 参考 (V* 4) [PutRecord](https://docs.aws.amazon.com/powershell/v4/reference)中的。
**适用于 PowerShell V5 的工具**
**示例 1：写入一条包含提供给 -Text 参数的字符串的记录。**
```
Write-KINRecord -Text "test data from string" -StreamName "mystream" -PartitionKey "Key1"
```
**示例 2：写入包含指定文件所包含数据的记录。该文件被视为字节序列，因此，如果它包含文本，则在将其与此 cmdlet 搭配使用之前，应使用任何必要的编码进行编写。**
```
Write-KINRecord -FilePath "C:\TestData.txt" -StreamName "mystream" -PartitionKey "Key2"
```
+ 有关 API 的详细信息，请参阅 *AWS Tools for PowerShell Cmdlet 参考 (V* 5) [PutRecord](https://docs.aws.amazon.com/powershell/v5/reference)中的。
------
#### [ Python ]
**适用于 Python 的 SDK（Boto3）**
还有更多相关信息 GitHub。在 [AWS 代码示例存储库](https://github.com/awsdocs/aws-doc-sdk-examples/tree/main/python/example_code/kinesis#code-examples)中查找完整示例，了解如何进行设置和运行。
```
class KinesisStream:
"""Encapsulates a Kinesis stream."""
def __init__(self, kinesis_client):
"""
:param kinesis_client: A Boto3 Kinesis client.
"""
self.kinesis_client = kinesis_client
self.name = None
self.details = None
self.stream_exists_waiter = kinesis_client.get_waiter("stream_exists")
def put_record(self, data, partition_key):
"""
Puts data into the stream. The data is formatted as JSON before it is passed
to the stream.
:param data: The data to put in the stream.
:param partition_key: The partition key to use for the data.
:return: Metadata about the record, including its shard ID and sequence number.
"""
try:
response = self.kinesis_client.put_record(
StreamName=self.name, Data=json.dumps(data), PartitionKey=partition_key
)
logger.info("Put record in stream %s.", self.name)
except ClientError:
logger.exception("Couldn't put record in stream %s.", self.name)
raise
else:
return response
```
+ 有关 API 的详细信息，请参阅适用[PutRecord](https://docs.aws.amazon.com/goto/boto3/kinesis-2013-12-02/PutRecord)于 *Python 的AWS SDK (Boto3) API 参考*。
------
#### [ Rust ]
**适用于 Rust 的 SDK**
还有更多相关信息 GitHub。在 [AWS 代码示例存储库](https://github.com/awsdocs/aws-doc-sdk-examples/tree/main/rustv1/examples/kinesis#code-examples)中查找完整示例，了解如何进行设置和运行。
```
async fn add_record(client: &Client, stream: &str, key: &str, data: &str) -> Result<(), Error> {
let blob = Blob::new(data);
client
.put_record()
.data(blob)
.partition_key(key)
.stream_name(stream)
.send()
.await?;
println!("Put data into stream.");
Ok(())
}
```
+ 有关 API 的详细信息，请参阅适用[PutRecord](https://docs.rs/aws-sdk-kinesis/latest/aws_sdk_kinesis/client/struct.Client.html#method.put_record)于 *Rust 的AWS SDK API 参考*。
------
#### [ SAP ABAP ]
**适用于 SAP ABAP 的 SDK**
还有更多相关信息 GitHub。在 [AWS 代码示例存储库](https://github.com/awsdocs/aws-doc-sdk-examples/tree/main/sap-abap/services/kns#code-examples)中查找完整示例，了解如何进行设置和运行。
```
TRY.
oo_result = lo_kns->putrecord( " oo_result is returned for testing purposes. "
iv_streamname = iv_stream_name
iv_data = iv_data
iv_partitionkey = iv_partition_key ).
MESSAGE 'Record created.' TYPE 'I'.
CATCH /aws1/cx_knsinvalidargumentex.
MESSAGE 'The specified argument was not valid.' TYPE 'E'.
CATCH /aws1/cx_knskmsaccessdeniedex.
MESSAGE 'You do not have permission to perform this AWS KMS action.' TYPE 'E'.
CATCH /aws1/cx_knskmsdisabledex.
MESSAGE 'KMS key used is disabled.' TYPE 'E'.
CATCH /aws1/cx_knskmsinvalidstateex.
MESSAGE 'KMS key used is in an invalid state. ' TYPE 'E'.
CATCH /aws1/cx_knskmsnotfoundex.
MESSAGE 'KMS key used is not found.' TYPE 'E'.
CATCH /aws1/cx_knskmsoptinrequired.
MESSAGE 'KMS key option is required.' TYPE 'E'.
CATCH /aws1/cx_knskmsthrottlingex.
MESSAGE 'The rate of requests to AWS KMS is exceeding the request quotas.' TYPE 'E'.
CATCH /aws1/cx_knsprovthruputexcdex.
MESSAGE 'The request rate for the stream is too high, or the requested data is too large for the available throughput.' TYPE 'E'.
CATCH /aws1/cx_knsresourcenotfoundex.
MESSAGE 'Resource being accessed is not found.' TYPE 'E'.
ENDTRY.
```
+ 有关 API 的详细信息，请参阅适用[PutRecord](https://docs.aws.amazon.com/sdk-for-sap-abap/v1/api/latest/index.html)于 S *AP 的AWS SDK ABAP API 参考*。
------
# `PutRecords`与 AWS SDK 或 CLI 配合使用
以下代码示例演示如何使用 `PutRecords`。
------
#### [ CLI ]
**AWS CLI**
**将多条记录写入一个数据流中**
以下 `put-records` 示例在一次调用中使用指定的分区键写入一条数据记录，并使用不同的分区键写入另一条数据记录。
```
aws kinesis put-records \
--stream-name samplestream \
--records Data=blob1,PartitionKey=partitionkey1 Data=blob2,PartitionKey=partitionkey2
```
输出：
```
{
"FailedRecordCount": 0,
"Records": [
{
"SequenceNumber": "49600883331171471519674795588238531498465399900093808706",
"ShardId": "shardId-000000000004"
},
{
"SequenceNumber": "49600902273357540915989931256902715169698037101720764562",
"ShardId": "shardId-000000000009"
}
],
"EncryptionType": "KMS"
}
```
有关更多信息，请参阅[《亚马逊 Kinesis Data Streams 开发者指南》中的使用亚马逊 Kinesis Data Streams API 和 Java 开发工具包开发生产](https://docs.aws.amazon.com/streams/latest/dev/developing-producers-with-sdk.html)*者*。 AWS
+ 有关 API 的详细信息，请参阅*AWS CLI 命令参考[PutRecords](https://awscli.amazonaws.com/v2/documentation/api/latest/reference/kinesis/put-records.html)*中的。
------
#### [ JavaScript ]
**适用于 JavaScript (v3) 的软件开发工具包**
还有更多相关信息 GitHub。在 [AWS 代码示例存储库](https://github.com/awsdocs/aws-doc-sdk-examples/tree/main/javascriptv3/example_code/kinesis#code-examples)中查找完整示例，了解如何进行设置和运行。
```
import { PutRecordsCommand, KinesisClient } from "@aws-sdk/client-kinesis";
/**
* Put multiple records into a Kinesis stream.
* @param {{ streamArn: string }} config
*/
export const main = async ({ streamArn }) => {
const client = new KinesisClient({});
try {
await client.send(
new PutRecordsCommand({
StreamARN: streamArn,
Records: [
{
Data: new Uint8Array(),
/**
* Determines which shard in the stream the data record is assigned to.
* Partition keys are Unicode strings with a maximum length limit of 256
* characters for each key. Amazon Kinesis Data Streams uses the partition
* key as input to a hash function that maps the partition key and
* associated data to a specific shard.
*/
PartitionKey: "TEST_KEY",
},
{
Data: new Uint8Array(),
PartitionKey: "TEST_KEY",
},
],
}),
);
} catch (caught) {
if (caught instanceof Error) {
//
} else {
throw caught;
}
}
};
// Call function if run directly.
import { fileURLToPath } from "node:url";
import { parseArgs } from "node:util";
if (process.argv[1] === fileURLToPath(import.meta.url)) {
const options = {
streamArn: {
type: "string",
description: "The ARN of the stream.",
},
};
const { values } = parseArgs({ options });
main(values);
}
```
+ 有关 API 的详细信息，请参阅 *适用于 JavaScript 的 AWS SDK API 参考[PutRecords](https://docs.aws.amazon.com/AWSJavaScriptSDK/v3/latest/client/kinesis/command/PutRecordsCommand)*中的。
------
# `RegisterStreamConsumer`与 AWS SDK 或 CLI 配合使用
以下代码示例演示如何使用 `RegisterStreamConsumer`。
------
#### [ .NET ]
**适用于 .NET 的 SDK**
还有更多相关信息 GitHub。在 [AWS 代码示例存储库](https://github.com/awsdocs/aws-doc-sdk-examples/tree/main/dotnetv3/Kinesis#code-examples)中查找完整示例，了解如何进行设置和运行。
```
using System;
using System.Threading.Tasks;
using Amazon.Kinesis;
using Amazon.Kinesis.Model;
///
/// This example shows how to register a consumer to an Amazon Kinesis
/// stream.
///
public class RegisterConsumer
{
public static async Task Main()
{
IAmazonKinesis client = new AmazonKinesisClient();
string consumerName = "NEW_CONSUMER_NAME";
string streamARN = "arn:aws:kinesis:us-east-2:000000000000:stream/AmazonKinesisStream";
var consumer = await RegisterConsumerAsync(client, consumerName, streamARN);
if (consumer is not null)
{
Console.WriteLine($"{consumer.ConsumerName}");
}
}
///
/// Registers the consumer to a Kinesis stream.
///
/// The initialized Kinesis client object.
/// A string representing the consumer.
/// The ARN of the stream.
/// A Consumer object that contains information about the consumer.
public static async Task RegisterConsumerAsync(IAmazonKinesis client, string consumerName, string streamARN)
{
var request = new RegisterStreamConsumerRequest
{
ConsumerName = consumerName,
StreamARN = streamARN,
};
var response = await client.RegisterStreamConsumerAsync(request);
return response.Consumer;
}
}
```
+ 有关 API 的详细信息，请参阅 *适用于 .NET 的 AWS SDK API 参考[RegisterStreamConsumer](https://docs.aws.amazon.com/goto/DotNetSDKV3/kinesis-2013-12-02/RegisterStreamConsumer)*中的。
------
#### [ CLI ]
**AWS CLI**
**注册数据流消费端**
以下 `register-stream-consumer` 示例将名为 `KinesisConsumerApplication` 的消费端注册到指定数据流。
```
aws kinesis register-stream-consumer \
--stream-arn arn:aws:kinesis:us-west-2:012345678912:stream/samplestream \
--consumer-name KinesisConsumerApplication
```
输出：
```
{
"Consumer": {
"ConsumerName": "KinesisConsumerApplication",
"ConsumerARN": "arn:aws:kinesis:us-west-2: 123456789012:stream/samplestream/consumer/KinesisConsumerApplication:1572383852",
"ConsumerStatus": "CREATING",
"ConsumerCreationTimestamp": 1572383852.0
}
}
```
有关更多信息，请参阅《Amazon Kinesis Data Streams 开发人员指南》中的[使用 Kinesis Data Streams API 开发具有增强型扇出功能的消费端](https://docs.aws.amazon.com/streams/latest/dev/building-enhanced-consumers-api.html)**。
+ 有关 API 的详细信息，请参阅*AWS CLI 命令参考[RegisterStreamConsumer](https://awscli.amazonaws.com/v2/documentation/api/latest/reference/kinesis/register-stream-consumer.html)*中的。
------
#### [ SAP ABAP ]
**适用于 SAP ABAP 的 SDK**
还有更多相关信息 GitHub。在 [AWS 代码示例存储库](https://github.com/awsdocs/aws-doc-sdk-examples/tree/main/sap-abap/services/kns#code-examples)中查找完整示例，了解如何进行设置和运行。
```
TRY.
oo_result = lo_kns->registerstreamconsumer( " oo_result is returned for testing purposes. "
iv_streamarn = iv_stream_arn
iv_consumername = iv_consumer_name ).
MESSAGE 'Stream consumer registered.' TYPE 'I'.
CATCH /aws1/cx_knsinvalidargumentex.
MESSAGE 'The specified argument was not valid.' TYPE 'E'.
CATCH /aws1/cx_sgmresourcelimitexcd.
MESSAGE 'You have reached the limit on the number of resources.' TYPE 'E'.
CATCH /aws1/cx_sgmresourceinuse.
MESSAGE 'Resource being accessed is in use.' TYPE 'E'.
CATCH /aws1/cx_sgmresourcenotfound.
MESSAGE 'Resource being accessed is not found.' TYPE 'E'.
ENDTRY.
```
+ 有关 API 的详细信息，请参阅适用[RegisterStreamConsumer](https://docs.aws.amazon.com/sdk-for-sap-abap/v1/api/latest/index.html)于 S *AP 的AWS SDK ABAP API 参考*。
------
# Kinesis 无服务器示例
以下代码示例展示了如何将 Kinesis 与配合使用。 AWS SDKs
**Topics**
+ [通过 Kinesis 触发器调用 Lambda 函数](kinesis_example_serverless_Kinesis_Lambda_section.md)
+ [通过 Kinesis 触发器报告 Lambda 函数批处理项目失败](kinesis_example_serverless_Kinesis_Lambda_batch_item_failures_section.md)
# 通过 Kinesis 触发器调用 Lambda 函数
以下代码示例演示了如何实现一个 Lambda 函数，该函数接收通过接收来自 Kinesis 流的记录而触发的事件。该函数检索 Kinesis 有效负载，将 Base64 解码，并记录下记录内容。
------
#### [ .NET ]
**适用于 .NET 的 SDK**
还有更多相关信息 GitHub。在[无服务器示例](https://github.com/aws-samples/serverless-snippets/tree/main/integration-kinesis-to-lambda)存储库中查找完整示例，并了解如何进行设置和运行。
通过 .NET 将 Kinesis 事件与 Lambda 结合使用。
```
// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
// SPDX-License-Identifier: Apache-2.0
using System.Text;
using Amazon.Lambda.Core;
using Amazon.Lambda.KinesisEvents;
using AWS.Lambda.Powertools.Logging;
// Assembly attribute to enable the Lambda function's JSON input to be converted into a .NET class.
[assembly: LambdaSerializer(typeof(Amazon.Lambda.Serialization.SystemTextJson.DefaultLambdaJsonSerializer))]
namespace KinesisIntegrationSampleCode;
public class Function
{
// Powertools Logger requires an environment variables against your function
// POWERTOOLS_SERVICE_NAME
[Logging(LogEvent = true)]
public async Task FunctionHandler(KinesisEvent evnt, ILambdaContext context)
{
if (evnt.Records.Count == 0)
{
Logger.LogInformation("Empty Kinesis Event received");
return;
}
foreach (var record in evnt.Records)
{
try
{
Logger.LogInformation($"Processed Event with EventId: {record.EventId}");
string data = await GetRecordDataAsync(record.Kinesis, context);
Logger.LogInformation($"Data: {data}");
// TODO: Do interesting work based on the new data
}
catch (Exception ex)
{
Logger.LogError($"An error occurred {ex.Message}");
throw;
}
}
Logger.LogInformation($"Successfully processed {evnt.Records.Count} records.");
}
private async Task GetRecordDataAsync(KinesisEvent.Record record, ILambdaContext context)
{
byte[] bytes = record.Data.ToArray();
string data = Encoding.UTF8.GetString(bytes);
await Task.CompletedTask; //Placeholder for actual async work
return data;
}
}
```
------
#### [ Go ]
**适用于 Go 的 SDK V2**
还有更多相关信息 GitHub。在[无服务器示例](https://github.com/aws-samples/serverless-snippets/tree/main/integration-kinesis-to-lambda)存储库中查找完整示例，并了解如何进行设置和运行。
使用 Go 将 Kinesis 事件与 Lambda 结合使用。
```
// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
// SPDX-License-Identifier: Apache-2.0
package main
import (
"context"
"log"
"github.com/aws/aws-lambda-go/events"
"github.com/aws/aws-lambda-go/lambda"
)
func handler(ctx context.Context, kinesisEvent events.KinesisEvent) error {
if len(kinesisEvent.Records) == 0 {
log.Printf("empty Kinesis event received")
return nil
}
for _, record := range kinesisEvent.Records {
log.Printf("processed Kinesis event with EventId: %v", record.EventID)
recordDataBytes := record.Kinesis.Data
recordDataText := string(recordDataBytes)
log.Printf("record data: %v", recordDataText)
// TODO: Do interesting work based on the new data
}
log.Printf("successfully processed %v records", len(kinesisEvent.Records))
return nil
}
func main() {
lambda.Start(handler)
}
```
------
#### [ Java ]
**适用于 Java 的 SDK 2.x**
还有更多相关信息 GitHub。在[无服务器示例](https://github.com/aws-samples/serverless-snippets/tree/main/integration-kinesis-to-lambda)存储库中查找完整示例，并了解如何进行设置和运行。
使用 Java 将 Kinesis 事件与 Lambda 结合使用。
```
// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
// SPDX-License-Identifier: Apache-2.0
package example;
import com.amazonaws.services.lambda.runtime.Context;
import com.amazonaws.services.lambda.runtime.LambdaLogger;
import com.amazonaws.services.lambda.runtime.RequestHandler;
import com.amazonaws.services.lambda.runtime.events.KinesisEvent;
public class Handler implements RequestHandler {
@Override
public Void handleRequest(final KinesisEvent event, final Context context) {
LambdaLogger logger = context.getLogger();
if (event.getRecords().isEmpty()) {
logger.log("Empty Kinesis Event received");
return null;
}
for (KinesisEvent.KinesisEventRecord record : event.getRecords()) {
try {
logger.log("Processed Event with EventId: "+record.getEventID());
String data = new String(record.getKinesis().getData().array());
logger.log("Data:"+ data);
// TODO: Do interesting work based on the new data
}
catch (Exception ex) {
logger.log("An error occurred:"+ex.getMessage());
throw ex;
}
}
logger.log("Successfully processed:"+event.getRecords().size()+" records");
return null;
}
}
```
------
#### [ JavaScript ]
**适用于 JavaScript (v3) 的软件开发工具包**
还有更多相关信息 GitHub。在[无服务器示例](https://github.com/aws-samples/serverless-snippets/blob/main/integration-kinesis-to-lambda)存储库中查找完整示例，并了解如何进行设置和运行。
使用 Lambda 消耗 Kinesis 事件。 JavaScript
```
// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
// SPDX-License-Identifier: Apache-2.0
exports.handler = async (event, context) => {
for (const record of event.Records) {
try {
console.log(`Processed Kinesis Event - EventID: ${record.eventID}`);
const recordData = await getRecordDataAsync(record.kinesis);
console.log(`Record Data: ${recordData}`);
// TODO: Do interesting work based on the new data
} catch (err) {
console.error(`An error occurred ${err}`);
throw err;
}
}
console.log(`Successfully processed ${event.Records.length} records.`);
};
async function getRecordDataAsync(payload) {
var data = Buffer.from(payload.data, "base64").toString("utf-8");
await Promise.resolve(1); //Placeholder for actual async work
return data;
}
```
使用 Lambda 消耗 Kinesis 事件。 TypeScript
```
// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
// SPDX-License-Identifier: Apache-2.0
import {
KinesisStreamEvent,
Context,
KinesisStreamHandler,
KinesisStreamRecordPayload,
} from "aws-lambda";
import { Buffer } from "buffer";
import { Logger } from "@aws-lambda-powertools/logger";
const logger = new Logger({
logLevel: "INFO",
serviceName: "kinesis-stream-handler-sample",
});
export const functionHandler: KinesisStreamHandler = async (
event: KinesisStreamEvent,
context: Context
): Promise => {
for (const record of event.Records) {
try {
logger.info(`Processed Kinesis Event - EventID: ${record.eventID}`);
const recordData = await getRecordDataAsync(record.kinesis);
logger.info(`Record Data: ${recordData}`);
// TODO: Do interesting work based on the new data
} catch (err) {
logger.error(`An error occurred ${err}`);
throw err;
}
logger.info(`Successfully processed ${event.Records.length} records.`);
}
};
async function getRecordDataAsync(
payload: KinesisStreamRecordPayload
): Promise {
var data = Buffer.from(payload.data, "base64").toString("utf-8");
await Promise.resolve(1); //Placeholder for actual async work
return data;
}
```
------
#### [ PHP ]
**适用于 PHP 的 SDK**
还有更多相关信息 GitHub。在[无服务器示例](https://github.com/aws-samples/serverless-snippets/tree/main/integration-kinesis-to-lambda)存储库中查找完整示例，并了解如何进行设置和运行。
通过 PHP 将 Kinesis 事件与 Lambda 结合使用。
```
// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
// SPDX-License-Identifier: Apache-2.0
logger = $logger;
}
/**
* @throws JsonException
* @throws \Bref\Event\InvalidLambdaEvent
*/
public function handleKinesis(KinesisEvent $event, Context $context): void
{
$this->logger->info("Processing records");
$records = $event->getRecords();
foreach ($records as $record) {
$data = $record->getData();
$this->logger->info(json_encode($data));
// TODO: Do interesting work based on the new data
// Any exception thrown will be logged and the invocation will be marked as failed
}
$totalRecords = count($records);
$this->logger->info("Successfully processed $totalRecords records");
}
}
$logger = new StderrLogger();
return new Handler($logger);
```
------
#### [ Python ]
**适用于 Python 的 SDK（Boto3）**
还有更多相关信息 GitHub。在[无服务器示例](https://github.com/aws-samples/serverless-snippets/tree/main/integration-kinesis-to-lambda)存储库中查找完整示例，并了解如何进行设置和运行。
使用 Python 将 Kinesis 事件与 Lambda 结合使用。
```
# Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
# SPDX-License-Identifier: Apache-2.0
import base64
def lambda_handler(event, context):
for record in event['Records']:
try:
print(f"Processed Kinesis Event - EventID: {record['eventID']}")
record_data = base64.b64decode(record['kinesis']['data']).decode('utf-8')
print(f"Record Data: {record_data}")
# TODO: Do interesting work based on the new data
except Exception as e:
print(f"An error occurred {e}")
raise e
print(f"Successfully processed {len(event['Records'])} records.")
```
------
#### [ Ruby ]
**适用于 Ruby 的 SDK**
还有更多相关信息 GitHub。在[无服务器示例](https://github.com/aws-samples/serverless-snippets/tree/main/integration-kinesis-to-lambda)存储库中查找完整示例，并了解如何进行设置和运行。
通过 Ruby 将 Kinesis 事件与 Lambda 结合使用。
```
# Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
# SPDX-License-Identifier: Apache-2.0
require 'aws-sdk'
def lambda_handler(event:, context:)
event['Records'].each do |record|
begin
puts "Processed Kinesis Event - EventID: #{record['eventID']}"
record_data = get_record_data_async(record['kinesis'])
puts "Record Data: #{record_data}"
# TODO: Do interesting work based on the new data
rescue => err
$stderr.puts "An error occurred #{err}"
raise err
end
end
puts "Successfully processed #{event['Records'].length} records."
end
def get_record_data_async(payload)
data = Base64.decode64(payload['data']).force_encoding('UTF-8')
# Placeholder for actual async work
# You can use Ruby's asynchronous programming tools like async/await or fibers here.
return data
end
```
------
#### [ Rust ]
**适用于 Rust 的 SDK**
还有更多相关信息 GitHub。在[无服务器示例](https://github.com/aws-samples/serverless-snippets/tree/main/integration-kinesis-to-lambda)存储库中查找完整示例，并了解如何进行设置和运行。
通过 Rust 将 Kinesis 事件与 Lambda 结合使用。
```
// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
// SPDX-License-Identifier: Apache-2.0
use aws_lambda_events::event::kinesis::KinesisEvent;
use lambda_runtime::{run, service_fn, Error, LambdaEvent};
async fn function_handler(event: LambdaEvent) -> Result<(), Error> {
if event.payload.records.is_empty() {
tracing::info!("No records found. Exiting.");
return Ok(());
}
event.payload.records.iter().for_each(|record| {
tracing::info!("EventId: {}",record.event_id.as_deref().unwrap_or_default());
let record_data = std::str::from_utf8(&record.kinesis.data);
match record_data {
Ok(data) => {
// log the record data
tracing::info!("Data: {}", data);
}
Err(e) => {
tracing::error!("Error: {}", e);
}
}
});
tracing::info!(
"Successfully processed {} records",
event.payload.records.len()
);
Ok(())
}
#[tokio::main]
async fn main() -> Result<(), Error> {
tracing_subscriber::fmt()
.with_max_level(tracing::Level::INFO)
// disable printing the name of the module in every log line.
.with_target(false)
// disabling time is handy because CloudWatch will add the ingestion time.
.without_time()
.init();
run(service_fn(function_handler)).await
}
```
------
# 通过 Kinesis 触发器报告 Lambda 函数批处理项目失败
以下代码示例显示如何为接收来自 Kinesis 流的事件的 Lambda 函数实现部分批处理响应。该函数在响应中报告批处理项目失败，并指示 Lambda 稍后重试这些消息。
------
#### [ .NET ]
**适用于 .NET 的 SDK**
还有更多相关信息 GitHub。在[无服务器示例](https://github.com/aws-samples/serverless-snippets/tree/main/integration-kinesis-to-lambda-with-batch-item-handling)存储库中查找完整示例，并了解如何进行设置和运行。
报告使用 .NET 进行 Lambda Kinesis 批处理项目失败。
```
// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
// SPDX-License-Identifier: Apache-2.0
using System.Text;
using System.Text.Json.Serialization;
using Amazon.Lambda.Core;
using Amazon.Lambda.KinesisEvents;
using AWS.Lambda.Powertools.Logging;
// Assembly attribute to enable the Lambda function's JSON input to be converted into a .NET class.
[assembly: LambdaSerializer(typeof(Amazon.Lambda.Serialization.SystemTextJson.DefaultLambdaJsonSerializer))]
namespace KinesisIntegration;
public class Function
{
// Powertools Logger requires an environment variables against your function
// POWERTOOLS_SERVICE_NAME
[Logging(LogEvent = true)]
public async Task FunctionHandler(KinesisEvent evnt, ILambdaContext context)
{
if (evnt.Records.Count == 0)
{
Logger.LogInformation("Empty Kinesis Event received");
return new StreamsEventResponse();
}
foreach (var record in evnt.Records)
{
try
{
Logger.LogInformation($"Processed Event with EventId: {record.EventId}");
string data = await GetRecordDataAsync(record.Kinesis, context);
Logger.LogInformation($"Data: {data}");
// TODO: Do interesting work based on the new data
}
catch (Exception ex)
{
Logger.LogError($"An error occurred {ex.Message}");
/* Since we are working with streams, we can return the failed item immediately.
Lambda will immediately begin to retry processing from this failed item onwards. */
return new StreamsEventResponse
{
BatchItemFailures = new List
{
new StreamsEventResponse.BatchItemFailure { ItemIdentifier = record.Kinesis.SequenceNumber }
}
};
}
}
Logger.LogInformation($"Successfully processed {evnt.Records.Count} records.");
return new StreamsEventResponse();
}
private async Task GetRecordDataAsync(KinesisEvent.Record record, ILambdaContext context)
{
byte[] bytes = record.Data.ToArray();
string data = Encoding.UTF8.GetString(bytes);
await Task.CompletedTask; //Placeholder for actual async work
return data;
}
}
public class StreamsEventResponse
{
[JsonPropertyName("batchItemFailures")]
public IList BatchItemFailures { get; set; }
public class BatchItemFailure
{
[JsonPropertyName("itemIdentifier")]
public string ItemIdentifier { get; set; }
}
}
```
------
#### [ Go ]
**适用于 Go 的 SDK V2**
还有更多相关信息 GitHub。在[无服务器示例](https://github.com/aws-samples/serverless-snippets/tree/main/integration-kinesis-to-lambda-with-batch-item-handling)存储库中查找完整示例，并了解如何进行设置和运行。
报告通过 Go 进行 Lambda Kinesis 批处理项目失败。
```
// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
// SPDX-License-Identifier: Apache-2.0
package main
import (
"context"
"fmt"
"github.com/aws/aws-lambda-go/events"
"github.com/aws/aws-lambda-go/lambda"
)
func handler(ctx context.Context, kinesisEvent events.KinesisEvent) (map[string]interface{}, error) {
batchItemFailures := []map[string]interface{}{}
for _, record := range kinesisEvent.Records {
curRecordSequenceNumber := ""
// Process your record
if /* Your record processing condition here */ {
curRecordSequenceNumber = record.Kinesis.SequenceNumber
}
// Add a condition to check if the record processing failed
if curRecordSequenceNumber != "" {
batchItemFailures = append(batchItemFailures, map[string]interface{}{"itemIdentifier": curRecordSequenceNumber})
}
}
kinesisBatchResponse := map[string]interface{}{
"batchItemFailures": batchItemFailures,
}
return kinesisBatchResponse, nil
}
func main() {
lambda.Start(handler)
}
```
------
#### [ Java ]
**适用于 Java 的 SDK 2.x**
还有更多相关信息 GitHub。在[无服务器示例](https://github.com/aws-samples/serverless-snippets/tree/main/integration-kinesis-to-lambda-with-batch-item-handling)存储库中查找完整示例，并了解如何进行设置和运行。
报告使用 Java 进行 Lambda Kinesis 批处理项目失败。
```
// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
// SPDX-License-Identifier: Apache-2.0
import com.amazonaws.services.lambda.runtime.Context;
import com.amazonaws.services.lambda.runtime.RequestHandler;
import com.amazonaws.services.lambda.runtime.events.KinesisEvent;
import com.amazonaws.services.lambda.runtime.events.StreamsEventResponse;
import java.io.Serializable;
import java.util.ArrayList;
import java.util.List;
public class ProcessKinesisRecords implements RequestHandler {
@Override
public StreamsEventResponse handleRequest(KinesisEvent input, Context context) {
List batchItemFailures = new ArrayList<>();
String curRecordSequenceNumber = "";
for (KinesisEvent.KinesisEventRecord kinesisEventRecord : input.getRecords()) {
try {
//Process your record
KinesisEvent.Record kinesisRecord = kinesisEventRecord.getKinesis();
curRecordSequenceNumber = kinesisRecord.getSequenceNumber();
} catch (Exception e) {
/* Since we are working with streams, we can return the failed item immediately.
Lambda will immediately begin to retry processing from this failed item onwards. */
batchItemFailures.add(new StreamsEventResponse.BatchItemFailure(curRecordSequenceNumber));
return new StreamsEventResponse(batchItemFailures);
}
}
return new StreamsEventResponse(batchItemFailures);
}
}
```
------
#### [ JavaScript ]
**适用于 JavaScript (v3) 的软件开发工具包**
还有更多相关信息 GitHub。在[无服务器示例](https://github.com/aws-samples/serverless-snippets/blob/main/integration-kinesis-to-lambda-with-batch-item-handling)存储库中查找完整示例，并了解如何进行设置和运行。
报告使用 Javascript 进行 Lambda Kinesis 批处理项目失败。
```
// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
// SPDX-License-Identifier: Apache-2.0
exports.handler = async (event, context) => {
for (const record of event.Records) {
try {
console.log(`Processed Kinesis Event - EventID: ${record.eventID}`);
const recordData = await getRecordDataAsync(record.kinesis);
console.log(`Record Data: ${recordData}`);
// TODO: Do interesting work based on the new data
} catch (err) {
console.error(`An error occurred ${err}`);
/* Since we are working with streams, we can return the failed item immediately.
Lambda will immediately begin to retry processing from this failed item onwards. */
return {
batchItemFailures: [{ itemIdentifier: record.kinesis.sequenceNumber }],
};
}
}
console.log(`Successfully processed ${event.Records.length} records.`);
return { batchItemFailures: [] };
};
async function getRecordDataAsync(payload) {
var data = Buffer.from(payload.data, "base64").toString("utf-8");
await Promise.resolve(1); //Placeholder for actual async work
return data;
}
```
使用 Lambda 报告 Kinesis 批处理项目失败。 TypeScript
```
// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
// SPDX-License-Identifier: Apache-2.0
import {
KinesisStreamEvent,
Context,
KinesisStreamHandler,
KinesisStreamRecordPayload,
KinesisStreamBatchResponse,
} from "aws-lambda";
import { Buffer } from "buffer";
import { Logger } from "@aws-lambda-powertools/logger";
const logger = new Logger({
logLevel: "INFO",
serviceName: "kinesis-stream-handler-sample",
});
export const functionHandler: KinesisStreamHandler = async (
event: KinesisStreamEvent,
context: Context
): Promise => {
for (const record of event.Records) {
try {
logger.info(`Processed Kinesis Event - EventID: ${record.eventID}`);
const recordData = await getRecordDataAsync(record.kinesis);
logger.info(`Record Data: ${recordData}`);
// TODO: Do interesting work based on the new data
} catch (err) {
logger.error(`An error occurred ${err}`);
/* Since we are working with streams, we can return the failed item immediately.
Lambda will immediately begin to retry processing from this failed item onwards. */
return {
batchItemFailures: [{ itemIdentifier: record.kinesis.sequenceNumber }],
};
}
}
logger.info(`Successfully processed ${event.Records.length} records.`);
return { batchItemFailures: [] };
};
async function getRecordDataAsync(
payload: KinesisStreamRecordPayload
): Promise {
var data = Buffer.from(payload.data, "base64").toString("utf-8");
await Promise.resolve(1); //Placeholder for actual async work
return data;
}
```
------
#### [ PHP ]
**适用于 PHP 的 SDK**
还有更多相关信息 GitHub。在[无服务器示例](https://github.com/aws-samples/serverless-snippets/tree/main/integration-kinesis-to-lambda-with-batch-item-handling)存储库中查找完整示例，并了解如何进行设置和运行。
报告通过 PHP 进行 Lambda Kinesis 批处理项目失败。
```
// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
// SPDX-License-Identifier: Apache-2.0
logger = $logger;
}
/**
* @throws JsonException
* @throws \Bref\Event\InvalidLambdaEvent
*/
public function handle(mixed $event, Context $context): array
{
$kinesisEvent = new KinesisEvent($event);
$this->logger->info("Processing records");
$records = $kinesisEvent->getRecords();
$failedRecords = [];
foreach ($records as $record) {
try {
$data = $record->getData();
$this->logger->info(json_encode($data));
// TODO: Do interesting work based on the new data
} catch (Exception $e) {
$this->logger->error($e->getMessage());
// failed processing the record
$failedRecords[] = $record->getSequenceNumber();
}
}
$totalRecords = count($records);
$this->logger->info("Successfully processed $totalRecords records");
// change format for the response
$failures = array_map(
fn(string $sequenceNumber) => ['itemIdentifier' => $sequenceNumber],
$failedRecords
);
return [
'batchItemFailures' => $failures
];
}
}
$logger = new StderrLogger();
return new Handler($logger);
```
------
#### [ Python ]
**适用于 Python 的 SDK（Boto3）**
还有更多相关信息 GitHub。在[无服务器示例](https://github.com/aws-samples/serverless-snippets/tree/main/integration-kinesis-to-lambda-with-batch-item-handling)存储库中查找完整示例，并了解如何进行设置和运行。
报告使用 Python 进行 Lambda Kinesis 批处理项目失败。
```
# Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
# SPDX-License-Identifier: Apache-2.0
def handler(event, context):
records = event.get("Records")
curRecordSequenceNumber = ""
for record in records:
try:
# Process your record
curRecordSequenceNumber = record["kinesis"]["sequenceNumber"]
except Exception as e:
# Return failed record's sequence number
return {"batchItemFailures":[{"itemIdentifier": curRecordSequenceNumber}]}
return {"batchItemFailures":[]}
```
------
#### [ Ruby ]
**适用于 Ruby 的 SDK**
还有更多相关信息 GitHub。在[无服务器示例](https://github.com/aws-samples/serverless-snippets/tree/main/integration-kinesis-to-lambda-with-batch-item-handling)存储库中查找完整示例，并了解如何进行设置和运行。
报告通过 Ruby 进行 Lambda Kinesis 批处理项目失败。
```
# Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
# SPDX-License-Identifier: Apache-2.0
require 'aws-sdk'
def lambda_handler(event:, context:)
batch_item_failures = []
event['Records'].each do |record|
begin
puts "Processed Kinesis Event - EventID: #{record['eventID']}"
record_data = get_record_data_async(record['kinesis'])
puts "Record Data: #{record_data}"
# TODO: Do interesting work based on the new data
rescue StandardError => err
puts "An error occurred #{err}"
# Since we are working with streams, we can return the failed item immediately.
# Lambda will immediately begin to retry processing from this failed item onwards.
return { batchItemFailures: [{ itemIdentifier: record['kinesis']['sequenceNumber'] }] }
end
end
puts "Successfully processed #{event['Records'].length} records."
{ batchItemFailures: batch_item_failures }
end
def get_record_data_async(payload)
data = Base64.decode64(payload['data']).force_encoding('utf-8')
# Placeholder for actual async work
sleep(1)
data
end
```
------
#### [ Rust ]
**适用于 Rust 的 SDK**
还有更多相关信息 GitHub。在[无服务器示例](https://github.com/aws-samples/serverless-snippets/tree/main/integration-kinesis-to-lambda-with-batch-item-handling)存储库中查找完整示例，并了解如何进行设置和运行。
报告通过 Rust 进行 Lambda Kinesis 批处理项目失败。
```
// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
// SPDX-License-Identifier: Apache-2.0
use aws_lambda_events::{
event::kinesis::KinesisEvent,
kinesis::KinesisEventRecord,
streams::{KinesisBatchItemFailure, KinesisEventResponse},
};
use lambda_runtime::{run, service_fn, Error, LambdaEvent};
async fn function_handler(event: LambdaEvent) -> Result {
let mut response = KinesisEventResponse {
batch_item_failures: vec![],
};
if event.payload.records.is_empty() {
tracing::info!("No records found. Exiting.");
return Ok(response);
}
for record in &event.payload.records {
tracing::info!(
"EventId: {}",
record.event_id.as_deref().unwrap_or_default()
);
let record_processing_result = process_record(record);
if record_processing_result.is_err() {
response.batch_item_failures.push(KinesisBatchItemFailure {
item_identifier: record.kinesis.sequence_number.clone(),
});
/* Since we are working with streams, we can return the failed item immediately.
Lambda will immediately begin to retry processing from this failed item onwards. */
return Ok(response);
}
}
tracing::info!(
"Successfully processed {} records",
event.payload.records.len()
);
Ok(response)
}
fn process_record(record: &KinesisEventRecord) -> Result<(), Error> {
let record_data = std::str::from_utf8(record.kinesis.data.as_slice());
if let Some(err) = record_data.err() {
tracing::error!("Error: {}", err);
return Err(Error::from(err));
}
let record_data = record_data.unwrap_or_default();
// do something interesting with the data
tracing::info!("Data: {}", record_data);
Ok(())
}
#[tokio::main]
async fn main() -> Result<(), Error> {
tracing_subscriber::fmt()
.with_max_level(tracing::Level::INFO)
// disable printing the name of the module in every log line.
.with_target(false)
// disabling time is handy because CloudWatch will add the ingestion time.
.without_time()
.init();
run(service_fn(function_handler)).await
}
```
------