Doc AWS SDK Examples GitHub リポジトリには、他にも SDK の例があります。 [AWS](https://github.com/awsdocs/aws-doc-sdk-examples)
翻訳は機械翻訳により提供されています。提供された翻訳内容と英語版の間で齟齬、不一致または矛盾がある場合、英語版が優先します。
# を使用した Kinesis の例 SDK for .NET
次のコード例は、Kinesis AWS SDK for .NET で を使用してアクションを実行し、一般的なシナリオを実装する方法を示しています。
*アクション*はより大きなプログラムからのコードの抜粋であり、コンテキスト内で実行する必要があります。アクションは個々のサービス機能を呼び出す方法を示していますが、コンテキスト内のアクションは、関連するシナリオで確認できます。
各例には完全なソースコードへのリンクが含まれており、コードの設定方法と実行方法に関する手順を確認できます。
**Topics**
+ [アクション](#actions)
+ [サーバーレスサンプル](#serverless_examples)
## アクション
### `AddTagsToStream`
次のコード例は、`AddTagsToStream` を使用する方法を示しています。
**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)」を参照してください。
### `CreateStream`
次のコード例は、`CreateStream` を使用する方法を示しています。
**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)」を参照してください。
### `DeleteStream`
次のコード例は、`DeleteStream` を使用する方法を示しています。
**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)」を参照してください。
### `DeregisterStreamConsumer`
次のコード例は、`DeregisterStreamConsumer` を使用する方法を示しています。
**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)」を参照してください。
### `ListStreamConsumers`
次のコード例は、`ListStreamConsumers` を使用する方法を示しています。
**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)を参照してください。
### `ListStreams`
次のコード例は、`ListStreams` を使用する方法を示しています。
**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)」を参照してください。
### `ListTagsForStream`
次のコード例は、`ListTagsForStream` を使用する方法を示しています。
**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)」を参照してください。
### `RegisterStreamConsumer`
次のコード例は、`RegisterStreamConsumer` を使用する方法を示しています。
**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;
}
}
```
+ API の詳細については、*AWS SDK for .NET API リファレンス*の「[スRegisterStreamConsumer](https://docs.aws.amazon.com/goto/DotNetSDKV3/kinesis-2013-12-02/RegisterStreamConsumer)」を参照してください。
## サーバーレスサンプル
### Kinesis トリガーから Lambda 関数を呼び出す
次のコード例では、Kinesis ストリームからレコードを受信することによってトリガーされるイベントを受け取る、Lambda 関数の実装方法を示しています。この関数は Kinesis ペイロードを取得し、それを Base64 からデコードして、そのレコードの内容をログ記録します。
**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;
}
}
```
### Kinesis トリガーを使用した Lambda 関数でのバッチアイテム失敗のレポート
以下のコード例では、Kinesis ストリームからイベントを受け取る Lambda 関数のための、部分的なバッチレスポンスの実装方法を示しています。この関数は、レスポンスとしてバッチアイテムの失敗を報告し、対象のメッセージを後で再試行するよう Lambda に伝えます。
**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; }
}
}
```