文件 AWS 開發套件範例 GitHub 儲存庫中有更多可用的 [AWS SDK 範例](https://github.com/awsdocs/aws-doc-sdk-examples)。
本文為英文版的機器翻譯版本，如內容有任何歧義或不一致之處，概以英文版為準。
# Kinesis AWS SDKs的程式碼範例
下列程式碼範例示範如何使用 Amazon Kinesis 搭配 AWS 軟體開發套件 (SDK)。
*基本概念*是程式碼範例，這些範例說明如何在服務內執行基本操作。
*Actions* 是大型程式的程式碼摘錄，必須在內容中執行。雖然動作會告訴您如何呼叫個別服務函數，但您可以在其相關情境中查看內容中的動作。
**其他資源**
+ **[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的基本範例
下列程式碼範例示範如何搭配 AWS SDK 使用 Amazon Kinesis 的基本功能。
**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)
# 使用 AWS SDK 了解 Kinesis 的基本概念
以下程式碼範例顯示做法：
+ 建立串流，並將記錄放入其中。
+ 建立碎片迭代器。
+ 讀取記錄，然後清除資源。
------
#### [ SAP ABAP ]
**適用於 SAP ABAP 的開發套件**
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 詳細資訊，請參閱《*適用於 SAP ABAP 的AWS SDK API 參考*》中的下列主題。
+ [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的動作
下列程式碼範例示範如何使用 AWS SDKs 執行個別 Kinesis 動作。每個範例均包含 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 ]
**SDK for Java 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 ]
**Tools for PowerShell V4**
**範例 1：建立新的串流。根據預設，此 Cmdlet 不會傳回任何輸出，因此會新增 -PassThru 切換變數，以傳回提供給 -StreamName 參數的值，供後續使用。**
```
$streamName = New-KINStream -StreamName "mystream" -ShardCount 1 -PassThru
```
+ 如需 API 詳細資訊，請參閱《AWS Tools for PowerShell Cmdlet 參考 (V4)》**中的 [CreateStream](https://docs.aws.amazon.com/powershell/v4/reference)。
**Tools for PowerShell V5**
**範例 1：建立新的串流。**
```
New-KINStream -StreamName "mystream" -ShardCount 1
```
+ 如需 API 詳細資訊，請參閱《AWS Tools for PowerShell Cmdlet 參考 (V5)》**中的 [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 詳細資訊，請參閱《AWS SDK for Python (Boto3) API 參考》**中的 [CreateStream](https://docs.aws.amazon.com/goto/boto3/kinesis-2013-12-02/CreateStream)。
------
#### [ 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 詳細資訊，請參閱《AWS SDK for Rust API 參考》**中的 [CreateStream](https://docs.rs/aws-sdk-kinesis/latest/aws_sdk_kinesis/client/struct.Client.html#method.create_stream)。
------
#### [ SAP ABAP ]
**適用於 SAP ABAP 的開發套件**
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 詳細資訊，請參閱《適用於 SAP ABAP 的AWS SDK API 參考》**中的 [CreateStream](https://docs.aws.amazon.com/sdk-for-sap-abap/v1/api/latest/index.html)。
------
# `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 ]
**SDK for Java 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 ]
**Tools for PowerShell V4**
**範例 1：刪除指定的串流。在執行命令之前，系統會提示您確認。若要隱藏確認提示，請使用 -Force 切換變數。**
```
Remove-KINStream -StreamName "mystream"
```
+ 如需 API 詳細資訊，請參閱《AWS Tools for PowerShell Cmdlet 參考 (V4)》**中的 [DeleteStream](https://docs.aws.amazon.com/powershell/v4/reference)。
**Tools for PowerShell V5**
**範例 1：刪除指定的串流。在執行命令之前，系統會提示您確認。若要隱藏確認提示，請使用 -Force 切換變數。**
```
Remove-KINStream -StreamName "mystream"
```
+ 如需 API 詳細資訊，請參閱《AWS Tools for PowerShell Cmdlet 參考 (V5)》**中的 [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 詳細資訊，請參閱《AWS SDK for Python (Boto3) API 參考》**中的 [DeleteStream](https://docs.aws.amazon.com/goto/boto3/kinesis-2013-12-02/DeleteStream)。
------
#### [ 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 詳細資訊，請參閱《AWS SDK for Rust API 參考》**中的 [DeleteStream](https://docs.rs/aws-sdk-kinesis/latest/aws_sdk_kinesis/client/struct.Client.html#method.delete_stream)。
------
#### [ SAP ABAP ]
**適用於 SAP ABAP 的開發套件**
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 詳細資訊，請參閱《適用於 SAP ABAP 的AWS SDK API 參考》**中的 [DeleteStream](https://docs.aws.amazon.com/sdk-for-sap-abap/v1/api/latest/index.html)。
------
# `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 ]
**Tools for 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 參考 (V4)》**中的 [DescribeStream](https://docs.aws.amazon.com/powershell/v4/reference)。
**Tools for 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 參考 (V5)》**中的 [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 詳細資訊，請參閱《AWS SDK for Python (Boto3) API 參考》**中的 [DescribeStream](https://docs.aws.amazon.com/goto/boto3/kinesis-2013-12-02/DescribeStream)。
------
#### [ 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 詳細資訊，請參閱《AWS SDK for Rust API 參考》**中的 [DescribeStream](https://docs.rs/aws-sdk-kinesis/latest/aws_sdk_kinesis/client/struct.Client.html#method.describe_stream)。
------
#### [ SAP ABAP ]
**適用於 SAP ABAP 的開發套件**
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 詳細資訊，請參閱《適用於 SAP ABAP 的 AWS SDK API 參考》**中的 [DescribeStream](https://docs.aws.amazon.com/sdk-for-sap-abap/v1/api/latest/index.html)。
------
# `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
}
```
如需詳細資訊，請參閱《Amazon [Kinesis Data Streams 開發人員指南》中的搭配適用於 Java 的 AWS SDK 使用 Kinesis Data Streams API 開發消費者](https://docs.aws.amazon.com/streams/latest/dev/developing-consumers-with-sdk.html)。 *Amazon Kinesis *
+ 如需 API 詳細資訊，請參閱《AWS CLI 命令參考》**中的 [GetRecords](https://awscli.amazonaws.com/v2/documentation/api/latest/reference/kinesis/get-records.html)。
------
#### [ Java ]
**SDK for Java 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 ]
**Tools for 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 參考 (V4)》**中的 [GetRecords](https://docs.aws.amazon.com/powershell/v4/reference)。
**Tools for 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 參考 (V5)》**中的 [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 詳細資訊，請參閱《AWS SDK for Python (Boto3) API 參考》**中的 [GetRecords](https://docs.aws.amazon.com/goto/boto3/kinesis-2013-12-02/GetRecords)。
------
#### [ SAP ABAP ]
**適用於 SAP ABAP 的開發套件**
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 詳細資訊，請參閱《適用於 SAP ABAP 的AWS SDK API 參考》**中的 [GetRecords](https://docs.aws.amazon.com/sdk-for-sap-abap/v1/api/latest/index.html)。
------
# 搭配使用 `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"
}
```
如需詳細資訊，請參閱《Amazon [Kinesis Data Streams 開發人員指南》中的搭配適用於 Java 的 AWS SDK 使用 Kinesis Data Streams API 開發消費者](https://docs.aws.amazon.com/streams/latest/dev/developing-consumers-with-sdk.html)。 *Amazon Kinesis *
+ 如需 API 詳細資訊，請參閱《AWS CLI 命令參考》**中的 [GetShardIterator](https://awscli.amazonaws.com/v2/documentation/api/latest/reference/kinesis/get-shard-iterator.html)。
------
#### [ PowerShell ]
**Tools for PowerShell V4**
**範例 1：傳回指定碎片和開始位置的碎片迭代器。碎片識別碼和序號的詳細資訊，可從 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 參考 (V4)》**中的 [GetShardIterator](https://docs.aws.amazon.com/powershell/v4/reference)。
**Tools for PowerShell V5**
**範例 1：傳回指定碎片和開始位置的碎片迭代器。碎片識別碼和序號的詳細資訊，可從 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 參考 (V5)》**中的 [GetShardIterator](https://docs.aws.amazon.com/powershell/v5/reference)。
------
# `ListStreamConsumers` 搭配 AWS SDK 使用
以下程式碼範例顯示如何使用 `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 詳細資訊，請參閱《AWS SDK for Rust API 參考》**中的 [ListStreams](https://docs.rs/aws-sdk-kinesis/latest/aws_sdk_kinesis/client/struct.Client.html#method.list_streams)。
------
#### [ SAP ABAP ]
**適用於 SAP ABAP 的開發套件**
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 詳細資訊，請參閱《適用於 SAP ABAP 的AWS SDK API 參考》**中的 [ListStreams](https://docs.aws.amazon.com/sdk-for-sap-abap/v1/api/latest/index.html)。
------
# `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"
}
```
如需詳細資訊，請參閱《[Amazon Kinesis Data Streams 開發人員指南》中的使用 Amazon Kinesis Data Streams API 搭配適用於 Java 的 AWS SDK 開發生產者](https://docs.aws.amazon.com/streams/latest/dev/developing-producers-with-sdk.html)。 *Amazon Kinesis *
+ 如需 API 詳細資訊，請參閱《AWS CLI 命令參考》**中的 [PutRecord](https://awscli.amazonaws.com/v2/documentation/api/latest/reference/kinesis/put-record.html)。
------
#### [ Java ]
**SDK for Java 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 ]
**Tools for 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 參考 (V4)》**中的 [PutRecord](https://docs.aws.amazon.com/powershell/v4/reference)。
**Tools for 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 參考 (V5)》**中的 [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 詳細資訊，請參閱《AWS SDK for Python (Boto3) API 參考》**中的 [PutRecord](https://docs.aws.amazon.com/goto/boto3/kinesis-2013-12-02/PutRecord)。
------
#### [ 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 詳細資訊，請參閱《AWS SDK for Rust API 參考》**中的 [PutRecord](https://docs.rs/aws-sdk-kinesis/latest/aws_sdk_kinesis/client/struct.Client.html#method.put_record)。
------
#### [ SAP ABAP ]
**適用於 SAP ABAP 的開發套件**
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 詳細資訊，請參閱《適用於 SAP ABAP 的AWS SDK API 參考》**中的 [PutRecord](https://docs.aws.amazon.com/sdk-for-sap-abap/v1/api/latest/index.html)。
------
# `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"
}
```
如需詳細資訊，請參閱《[Amazon Kinesis Data Streams 開發人員指南》中的使用 Amazon Kinesis Data Streams API 搭配適用於 Java 的 AWS SDK 開發生產者](https://docs.aws.amazon.com/streams/latest/dev/developing-producers-with-sdk.html)。 *Amazon Kinesis *
+ 如需 API 詳細資訊，請參閱《AWS CLI 命令參考》**中的 [PutRecords](https://awscli.amazonaws.com/v2/documentation/api/latest/reference/kinesis/put-records.html)。
------
#### [ JavaScript ]
**適用於 JavaScript (v3) 的 SDK**
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 的開發套件**
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 詳細資訊，請參閱《適用於 SAP ABAP 的 AWS SDK API 參考》**中的 [RegisterStreamConsumer](https://docs.aws.amazon.com/sdk-for-sap-abap/v1/api/latest/index.html)。
------
# 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 搭配 Lambda 來使用 Kinesis 事件。
```
// 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 ]
**SDK for Go V2**
GitHub 上提供更多範例。尋找完整範例，並了解如何在[無伺服器範例](https://github.com/aws-samples/serverless-snippets/tree/main/integration-kinesis-to-lambda)儲存庫中設定和執行。
使用 Go 搭配 Lambda 來使用 Kinesis 事件。
```
// 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 ]
**SDK for Java 2.x**
GitHub 上提供更多範例。尋找完整範例，並了解如何在[無伺服器範例](https://github.com/aws-samples/serverless-snippets/tree/main/integration-kinesis-to-lambda)儲存庫中設定和執行。
使用 Java 搭配 Lambda 來使用 Kinesis 事件。
```
// 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) 的 SDK**
GitHub 上提供更多範例。尋找完整範例，並了解如何在[無伺服器範例](https://github.com/aws-samples/serverless-snippets/blob/main/integration-kinesis-to-lambda)儲存庫中設定和執行。
使用 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}`);
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;
}
```
使用 TypeScript 搭配 Lambda 來使用 Kinesis 事件。
```
// 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 搭配 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 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 搭配 Lambda 來使用 Kinesis 事件。
```
# 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 ]
**SDK for Ruby**
GitHub 上提供更多範例。尋找完整範例，並了解如何在[無伺服器範例](https://github.com/aws-samples/serverless-snippets/tree/main/integration-kinesis-to-lambda)儲存庫中設定和執行。
使用 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:)
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 搭配 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;
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 ]
**SDK for Go 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 ]
**SDK for Java 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) 的 SDK**
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;
}
```
使用 TypeScript 搭配 Lambda 報告 Kinesis 批次項目失敗。
```
// 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 ]
**SDK for Ruby**
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
}
```
------