기계 번역으로 제공되는 번역입니다. 제공된 번역과 원본 영어의 내용이 상충하는 경우에는 영어 버전이 우선합니다.
# AWS SDKs를 사용한 Kinesis 코드 예제
다음 코드 예제에서는 Kinesis를 AWS 소프트웨어 개발 키트(SDK)와 함께 사용하는 방법을 보여줍니다.
*기본 사항*은 서비스 내에서 필수 작업을 수행하는 방법을 보여주는 코드 예제입니다.
*작업*은 대규모 프로그램에서 발췌한 코드이며 컨텍스트에 맞춰 실행해야 합니다. 작업은 개별 서비스 함수를 직접 호출하는 방법을 보여주며, 관련 시나리오의 컨텍스트에 맞는 작업을 볼 수 있습니다.
AWS SDK 개발자 안내서 및 코드 예제의 전체 목록은 섹션을 참조하세요[AWS SDK에서이 서비스 사용](sdk-general-information-section.md). 이 주제에는 시작하기에 대한 정보와 이전 SDK 버전에 대한 세부 정보도 포함되어 있습니다.
**Contents**
+ [기본 사항](service_code_examples_basics.md)
+ [기본 사항 알아보기](example_kinesis_Scenario_GettingStarted_section.md)
+ [작업](service_code_examples_actions.md)
+ [`AddTagsToStream`](example_kinesis_AddTagsToStream_section.md)
+ [`CreateStream`](example_kinesis_CreateStream_section.md)
+ [`DeleteStream`](example_kinesis_DeleteStream_section.md)
+ [`DeregisterStreamConsumer`](example_kinesis_DeregisterStreamConsumer_section.md)
+ [`DescribeStream`](example_kinesis_DescribeStream_section.md)
+ [`GetRecords`](example_kinesis_GetRecords_section.md)
+ [`GetShardIterator`](example_kinesis_GetShardIterator_section.md)
+ [`ListStreamConsumers`](example_kinesis_ListStreamConsumers_section.md)
+ [`ListStreams`](example_kinesis_ListStreams_section.md)
+ [`ListTagsForStream`](example_kinesis_ListTagsForStream_section.md)
+ [`PutRecord`](example_kinesis_PutRecord_section.md)
+ [`PutRecords`](example_kinesis_PutRecords_section.md)
+ [`RegisterStreamConsumer`](example_kinesis_RegisterStreamConsumer_section.md)
+ [서버리스 예제](service_code_examples_serverless_examples.md)
+ [Kinesis 트리거에서 간접적으로 Lambda 함수 간접 호출](example_serverless_Kinesis_Lambda_section.md)
+ [Kinesis 트리거로 Lambda 함수에 대한 배치 항목 실패 보고](example_serverless_Kinesis_Lambda_batch_item_failures_section.md)
# AWS SDKs 사용한 Kinesis의 기본 예제
다음 코드 예제에서는 AWS SDK와 함께 Amazon Kinesis의 기본 사항을 사용하는 방법을 보여줍니다.
**Contents**
+ [기본 사항 알아보기](example_kinesis_Scenario_GettingStarted_section.md)
+ [작업](service_code_examples_actions.md)
+ [`AddTagsToStream`](example_kinesis_AddTagsToStream_section.md)
+ [`CreateStream`](example_kinesis_CreateStream_section.md)
+ [`DeleteStream`](example_kinesis_DeleteStream_section.md)
+ [`DeregisterStreamConsumer`](example_kinesis_DeregisterStreamConsumer_section.md)
+ [`DescribeStream`](example_kinesis_DescribeStream_section.md)
+ [`GetRecords`](example_kinesis_GetRecords_section.md)
+ [`GetShardIterator`](example_kinesis_GetShardIterator_section.md)
+ [`ListStreamConsumers`](example_kinesis_ListStreamConsumers_section.md)
+ [`ListStreams`](example_kinesis_ListStreams_section.md)
+ [`ListTagsForStream`](example_kinesis_ListTagsForStream_section.md)
+ [`PutRecord`](example_kinesis_PutRecord_section.md)
+ [`PutRecords`](example_kinesis_PutRecords_section.md)
+ [`RegisterStreamConsumer`](example_kinesis_RegisterStreamConsumer_section.md)
# AWS SDK를 사용하여 Kinesis의 기본 사항 알아보기
다음 코드 예제에서는 다음과 같은 작업을 수행하는 방법을 보여줍니다.
+ 스트림을 생성하고 그 안에 레코드를 넣습니다.
+ 샤드 반복자를 생성합니다.
+ 레코드를 읽은 다음 리소스를 정리합니다.
------
#### [ SAP ABAP ]
**SDK for SAP ABAP API**
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 참조*의 다음 주제를 참조하세요.
+ [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)
------
AWS SDK 개발자 안내서 및 코드 예제의 전체 목록은 섹션을 참조하세요[AWS SDK에서이 서비스 사용](sdk-general-information-section.md). 이 주제에는 시작하기에 대한 정보와 이전 SDK 버전에 대한 세부 정보도 포함되어 있습니다.
# AWS SDKs를 사용한 Kinesis 작업
다음 코드 예제에서는 AWS SDKs를 사용하여 개별 Kinesis 작업을 수행하는 방법을 보여줍니다. 각 예시에는 GitHub에 대한 링크가 포함되어 있습니다. 여기에서 코드 설정 및 실행에 대한 지침을 찾을 수 있습니다.
다음 예제에는 가장 일반적으로 사용되는 작업만 포함되어 있습니다. 전체 목록은 [Amazon Kinesis API 참조](https://docs.aws.amazon.com/kinesis/latest/APIReference/Welcome.html)를 참조하세요.
**Topics**
+ [`AddTagsToStream`](example_kinesis_AddTagsToStream_section.md)
+ [`CreateStream`](example_kinesis_CreateStream_section.md)
+ [`DeleteStream`](example_kinesis_DeleteStream_section.md)
+ [`DeregisterStreamConsumer`](example_kinesis_DeregisterStreamConsumer_section.md)
+ [`DescribeStream`](example_kinesis_DescribeStream_section.md)
+ [`GetRecords`](example_kinesis_GetRecords_section.md)
+ [`GetShardIterator`](example_kinesis_GetShardIterator_section.md)
+ [`ListStreamConsumers`](example_kinesis_ListStreamConsumers_section.md)
+ [`ListStreams`](example_kinesis_ListStreams_section.md)
+ [`ListTagsForStream`](example_kinesis_ListTagsForStream_section.md)
+ [`PutRecord`](example_kinesis_PutRecord_section.md)
+ [`PutRecords`](example_kinesis_PutRecords_section.md)
+ [`RegisterStreamConsumer`](example_kinesis_RegisterStreamConsumer_section.md)
# AWS SDK 또는 CLI와 `AddTagsToStream` 함께 사용
다음 코드 예시는 `AddTagsToStream`의 사용 방법을 보여 줍니다.
------
#### [ .NET ]
**SDK for .NET**
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 세부 정보는 *AWS SDK for .NET 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)을 참조하세요.
------
AWS SDK 개발자 안내서 및 코드 예제의 전체 목록은 섹션을 참조하세요[AWS SDK에서이 서비스 사용](sdk-general-information-section.md). 이 주제에는 시작하기에 대한 정보와 이전 SDK 버전에 대한 세부 정보도 포함되어 있습니다.
# AWS SDK 또는 CLI와 `CreateStream` 함께 사용
다음 코드 예시는 `CreateStream`의 사용 방법을 보여 줍니다.
작업 예제는 대규모 프로그램에서 발췌한 코드이며 컨텍스트에 맞춰 실행해야 합니다. 다음 코드 예제에서는 컨텍스트 내에서 이 작업을 확인할 수 있습니다.
+ [기본 사항 알아보기](example_kinesis_Scenario_GettingStarted_section.md)
------
#### [ .NET ]
**SDK for .NET**
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에 대한 세부 정보는 *AWS SDK for .NET API 참조*의 [CreateStream](https://docs.aws.amazon.com/goto/DotNetSDKV3/kinesis-2013-12-02/CreateStream)을 참조하세요.
------
#### [ CLI ]
**AWS CLI**
**데이터 스트림을 생성하는 방법**
다음 `create-stream` 예시에서는 샤드 3개가 포함된 samplestream이라는 데이터 스트림을 생성합니다.
```
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은 출력을 반환하지 않으므로 이후에 사용할 수 있게 -StreamName 파라미터에 제공된 값을 반환하기 위해 -PassThru 스위치가 추가됩니다.**
```
$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 ]
**SDK for Python(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 ]
**SDK for Rust**
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 ]
**SDK for SAP ABAP API**
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에 대한 세부 정보는 *AWS SDK for SAP ABAP API 참조*의 [CreateStream](https://docs.aws.amazon.com/sdk-for-sap-abap/v1/api/latest/index.html)을 참조하세요.
------
AWS SDK 개발자 안내서 및 코드 예제의 전체 목록은 섹션을 참조하세요[AWS SDK에서이 서비스 사용](sdk-general-information-section.md). 이 주제에는 시작하기에 대한 정보와 이전 SDK 버전에 대한 세부 정보도 포함되어 있습니다.
# AWS SDK 또는 CLI와 `DeleteStream` 함께 사용
다음 코드 예시는 `DeleteStream`의 사용 방법을 보여 줍니다.
작업 예제는 대규모 프로그램에서 발췌한 코드이며 컨텍스트에 맞춰 실행해야 합니다. 다음 코드 예제에서는 컨텍스트 내에서 이 작업을 확인할 수 있습니다.
+ [기본 사항 알아보기](example_kinesis_Scenario_GettingStarted_section.md)
------
#### [ .NET ]
**SDK for .NET**
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에 대한 세부 정보는 *AWS SDK for .NET 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 ]
**SDK for Python(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 ]
**SDK for Rust**
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 ]
**SDK for SAP ABAP API**
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에 대한 세부 정보는 *AWS SDK for SAP ABAP API 참조*의 [DeleteStream](https://docs.aws.amazon.com/sdk-for-sap-abap/v1/api/latest/index.html)을 참조하세요.
------
AWS SDK 개발자 안내서 및 코드 예제의 전체 목록은 섹션을 참조하세요[AWS SDK에서이 서비스 사용](sdk-general-information-section.md). 이 주제에는 시작하기에 대한 정보와 이전 SDK 버전에 대한 세부 정보도 포함되어 있습니다.
# AWS SDK 또는 CLI와 `DeregisterStreamConsumer` 함께 사용
다음 코드 예시는 `DeregisterStreamConsumer`의 사용 방법을 보여 줍니다.
------
#### [ .NET ]
**SDK for .NET**
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 세부 정보는 *AWS SDK for .NET 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)를 참조하세요.
------
AWS SDK 개발자 안내서 및 코드 예제의 전체 목록은 섹션을 참조하세요[AWS SDK에서이 서비스 사용](sdk-general-information-section.md). 이 주제에는 시작하기에 대한 정보와 이전 SDK 버전에 대한 세부 정보도 포함되어 있습니다.
# AWS SDK 또는 CLI와 `DescribeStream` 함께 사용
다음 코드 예시는 `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 ]
**SDK for Python(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 ]
**SDK for Rust**
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 ]
**SDK for SAP ABAP API**
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에 대한 세부 정보는 *AWS SDK for SAP ABAP API 참조*의 [DescribeStream](https://docs.aws.amazon.com/sdk-for-sap-abap/v1/api/latest/index.html)을 참조하세요.
------
AWS SDK 개발자 안내서 및 코드 예제의 전체 목록은 섹션을 참조하세요[AWS SDK에서이 서비스 사용](sdk-general-information-section.md). 이 주제에는 시작하기에 대한 정보와 이전 SDK 버전에 대한 세부 정보도 포함되어 있습니다.
# AWS SDK 또는 CLI와 `GetRecords` 함께 사용
다음 코드 예시는 `GetRecords`의 사용 방법을 보여 줍니다.
작업 예제는 대규모 프로그램에서 발췌한 코드이며 컨텍스트에 맞춰 실행해야 합니다. 다음 코드 예제에서는 컨텍스트 내에서 이 작업을 확인할 수 있습니다.
+ [기본 사항 알아보기](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 개발자 안내서의 AWS SDK for Java와 함께 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 ]
**SDK for Python(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 ]
**SDK for SAP ABAP API**
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 세부 정보는 **AWS SDK for SAP ABAP API 참조의 [GetRecords](https://docs.aws.amazon.com/sdk-for-sap-abap/v1/api/latest/index.html)를 참조하세요.
------
AWS SDK 개발자 안내서 및 코드 예제의 전체 목록은 섹션을 참조하세요[AWS SDK에서이 서비스 사용](sdk-general-information-section.md). 이 주제에는 시작하기에 대한 정보와 이전 SDK 버전에 대한 세부 정보도 포함되어 있습니다.
# CLI로 `GetShardIterator` 사용
다음 코드 예시는 `GetShardIterator`의 사용 방법을 보여 줍니다.
작업 예제는 대규모 프로그램에서 발췌한 코드이며 컨텍스트에 맞춰 실행해야 합니다. 다음 코드 예제에서는 컨텍스트 내에서 이 작업을 확인할 수 있습니다.
+ [기본 사항 알아보기](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 개발자 안내서의 AWS SDK for Java와 함께 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)를 참조하세요.
------
AWS SDK 개발자 안내서 및 코드 예제의 전체 목록은 섹션을 참조하세요[AWS SDK에서이 서비스 사용](sdk-general-information-section.md). 이 주제에는 시작하기에 대한 정보와 이전 SDK 버전에 대한 세부 정보도 포함되어 있습니다.
# AWS SDK와 `ListStreamConsumers` 함께 사용
다음 코드 예시는 `ListStreamConsumers`의 사용 방법을 보여 줍니다.
------
#### [ .NET ]
**SDK for .NET**
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 세부 정보는 *AWS SDK for .NET API 참조*의 [ListStreamConsumers](https://docs.aws.amazon.com/goto/DotNetSDKV3/kinesis-2013-12-02/ListStreamConsumers)를 참조하세요.
------
AWS SDK 개발자 안내서 및 코드 예제의 전체 목록은 섹션을 참조하세요[AWS SDK에서이 서비스 사용](sdk-general-information-section.md). 이 주제에는 시작하기에 대한 정보와 이전 SDK 버전에 대한 세부 정보도 포함되어 있습니다.
# AWS SDK 또는 CLI와 `ListStreams` 함께 사용
다음 코드 예시는 `ListStreams`의 사용 방법을 보여 줍니다.
------
#### [ .NET ]
**SDK for .NET**
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 세부 정보는 *AWS SDK for .NET 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 ]
**SDK for Rust**
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 ]
**SDK for SAP ABAP API**
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에 대한 세부 정보는 *AWS SDK for SAP ABAP API 참조*의 [ListStreams](https://docs.aws.amazon.com/sdk-for-sap-abap/v1/api/latest/index.html)을 참조하세요.
------
AWS SDK 개발자 안내서 및 코드 예제의 전체 목록은 섹션을 참조하세요[AWS SDK에서이 서비스 사용](sdk-general-information-section.md). 이 주제에는 시작하기에 대한 정보와 이전 SDK 버전에 대한 세부 정보도 포함되어 있습니다.
# AWS SDK 또는 CLI와 `ListTagsForStream` 함께 사용
다음 코드 예시는 `ListTagsForStream`의 사용 방법을 보여 줍니다.
------
#### [ .NET ]
**SDK for .NET**
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에 대한 세부 정보는 *AWS SDK for .NET 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)을 참조하세요.
------
AWS SDK 개발자 안내서 및 코드 예제의 전체 목록은 섹션을 참조하세요[AWS SDK에서이 서비스 사용](sdk-general-information-section.md). 이 주제에는 시작하기에 대한 정보와 이전 SDK 버전에 대한 세부 정보도 포함되어 있습니다.
# AWS SDK 또는 CLI와 `PutRecord` 함께 사용
다음 코드 예시는 `PutRecord`의 사용 방법을 보여 줍니다.
작업 예제는 대규모 프로그램에서 발췌한 코드이며 컨텍스트에 맞춰 실행해야 합니다. 다음 코드 예제에서는 컨텍스트 내에서 이 작업을 확인할 수 있습니다.
+ [기본 사항 알아보기](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 개발자 안내서의 Java용 AWS SDK와 함께 Amazon Kinesis Data Streams API를 사용하여 생산자 개발을](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 ]
**SDK for Python(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 ]
**SDK for Rust**
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 ]
**SDK for SAP ABAP API**
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에 대한 세부 정보는 *AWS SDK for SAP ABAP API 참조*의 [PutRecord](https://docs.aws.amazon.com/sdk-for-sap-abap/v1/api/latest/index.html)를 참조하세요.
------
AWS SDK 개발자 안내서 및 코드 예제의 전체 목록은 섹션을 참조하세요[AWS SDK에서이 서비스 사용](sdk-general-information-section.md). 이 주제에는 시작하기에 대한 정보와 이전 SDK 버전에 대한 세부 정보도 포함되어 있습니다.
# AWS SDK 또는 CLI와 `PutRecords` 함께 사용
다음 코드 예시는 `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 개발자 안내서의 Java용 AWS SDK와 함께 Amazon Kinesis Data Streams API를 사용하여 생산자 개발을](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 ]
**SDK for 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에 대한 세부 정보는 **AWS SDK for JavaScript API 참조의 [PutRecords](https://docs.aws.amazon.com/AWSJavaScriptSDK/v3/latest/client/kinesis/command/PutRecordsCommand)를 참조하세요.
------
AWS SDK 개발자 안내서 및 코드 예제의 전체 목록은 섹션을 참조하세요[AWS SDK에서이 서비스 사용](sdk-general-information-section.md). 이 주제에는 시작하기에 대한 정보와 이전 SDK 버전에 대한 세부 정보도 포함되어 있습니다.
# AWS SDK 또는 CLI와 `RegisterStreamConsumer` 함께 사용
다음 코드 예시는 `RegisterStreamConsumer`의 사용 방법을 보여 줍니다.
------
#### [ .NET ]
**SDK for .NET**
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;
}
}
```
+ 자세한 내용은 *AWS SDK for .NET 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)을 참조하세요.
+ 자세한 내용은 **AWS CLI 명령 참조의 [RegisterStreamConsumer](https://awscli.amazonaws.com/v2/documentation/api/latest/reference/kinesis/register-stream-consumer.html)를 참조하세요.
------
#### [ SAP ABAP ]
**SDK for SAP ABAP API**
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에 대한 세부 정보는 *AWS SDK for SAP ABAP API 참조*의 [RegisterStreamConsumer](https://docs.aws.amazon.com/sdk-for-sap-abap/v1/api/latest/index.html)를 참조하세요.
------
AWS SDK 개발자 안내서 및 코드 예제의 전체 목록은 섹션을 참조하세요[AWS SDK에서이 서비스 사용](sdk-general-information-section.md). 이 주제에는 시작하기에 대한 정보와 이전 SDK 버전에 대한 세부 정보도 포함되어 있습니다.
# Kinesis의 서버리스 예제
다음 코드 예제에서는 Kinesis를 AWS SDKs 함께 사용하는 방법을 보여줍니다.
**Topics**
+ [Kinesis 트리거에서 간접적으로 Lambda 함수 간접 호출](example_serverless_Kinesis_Lambda_section.md)
+ [Kinesis 트리거로 Lambda 함수에 대한 배치 항목 실패 보고](example_serverless_Kinesis_Lambda_batch_item_failures_section.md)
# Kinesis 트리거에서 간접적으로 Lambda 함수 호출
다음 코드 예제에서는 Kinesis 스트림에서 레코드를 받아 트리거된 이벤트를 수신하는 Lambda 함수를 구현하는 방법을 보여줍니다. 이 함수는 Kinesis 페이로드를 검색하고, Base64에서 디코딩하고, 레코드 콘텐츠를 로깅합니다.
------
#### [ .NET ]
**SDK for .NET**
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 ]
**SDK for JavaScript (v3)**
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 ]
**SDK for PHP**
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 ]
**SDK for Python(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 ]
**SDK for Rust**
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
}
```
------
AWS SDK 개발자 안내서 및 코드 예제의 전체 목록은 섹션을 참조하세요[AWS SDK에서이 서비스 사용](sdk-general-information-section.md). 이 주제에는 시작하기에 대한 정보와 이전 SDK 버전에 대한 세부 정보도 포함되어 있습니다.
# Kinesis 트리거로 Lambda 함수에 대한 배치 항목 실패 보고
다음 코드 예제에서는 Kinesis 스트림에서 이벤트를 수신하는 Lambda 함수에 대한 부분 배치 응답을 구현하는 방법을 보여줍니다. 이 함수는 응답으로 배치 항목 실패를 보고하고 나중에 해당 메시지를 다시 시도하도록 Lambda에 신호를 보냅니다.
------
#### [ .NET ]
**SDK for .NET**
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 ]
**SDK for 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;
}
```
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 ]
**SDK for PHP**
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 ]
**SDK for Python(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 ]
**SDK for Rust**
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
}
```
------
AWS SDK 개발자 안내서 및 코드 예제의 전체 목록은 섹션을 참조하세요[AWS SDK에서이 서비스 사용](sdk-general-information-section.md). 이 주제에는 시작하기에 대한 정보와 이전 SDK 버전에 대한 세부 정보도 포함되어 있습니다.