KinesisでAPIGatewayを使用してS3にデータを送信してみた。

SUMANGALAS

2022.01.28

この記事は公開されてから1年以上経過しています。情報が古い可能性がありますので、ご注意ください。

概要

この記事では、Kinesis を介してS3にデータを送信するREST APIを作成してみました。APIはPutRecordアクションを使用してデータをKinesis Data Streamsに送信します。Kinesis Firehoseは、KinesisStreamsからS3バケットにデータを送信します。

次のように API を作成しておきました。

HTTP GET メソッド　: Resource - /streams, Action - ListStreams. このメソッドはアカウントのストリームを一覧表示します。
HTTP GET メソッド　: Resource - /streams/{stream-name}, Action - DescribeStreams. このメソッドは指定したストリームを表示します。
HTTP PUTメソッド　: Resource - /streams/{stream-name}/record, Action - PutRecord. このメソッドはストリームに 1 つのデータレコードを追加します。

やってみた

Kinesis Data Streamの作成

Amazon Kinesisコンソールで、Data Streamsを選択して、Create data streamをクリックしておきます。
次の設定でData Streamを作成しておきます。
- データストリーム名　: 　データストリームの名前を入力します。
- Data stream capacity　：　Provisioned
- Provisioned shards　：　１

S3 バケットの作成

S3 コンソールで、Create bucketを選択します。
バケット名を入力して、S3バケットを作成しておきます。

Kinesis Data Firehose Deliveryストリームの作成

Kinesis Data Firehoseのコンソールで、Create Delivery Streamを選択します。
SourceでAmazon Kinesis Data Streamsを選択して、 DestinationでAmason S3を選択しておきます。
Source settingsで以前に作成したデータストリームを選択して、Deliveryストリーム名入力しておきます。

Destination設定で以前に作成したS3バケットURLの名前を入力します。

Buffer intervalを60Secondsに設定して、Deliveryストリームを作成しておきます。

REST APIの作成

API GatewayコンソールのREST APIペインで、Buildを選択しておきます。

New API を選択して、API名を入力して、Endpoint typeで regionalを選択して、Create APIをクリックしておきます。

APIのリソースとメソッドの作成

IAM ロールの作成

API GatewayのIAMロールを作成しておきます。
この設定でIAMロールを作成しておきます。
- 信頼されたエンティティの種類　:　AWS のサービス
- ユースケース　:　API Gateway
- ポリシー : AmazonKinesisAnalyticsFullAccess
このIAMロールは、API Gatewayのメソッドを作成するときに使用されます。

List streamsのリソースとメソッド

Resources ページで、Actionsをクリック、Create Resourceを選択しておきます。
リソース名[streams]を入力して、[Create Resource]をクリックしておきます。

Actionsで、Create Methodを選択して、GETを選択しておきます。
次の設定でGET Methodを作成しておきます。
- Integration type : AWS Service
- Region : us-east-1
- AWS Service : Kinesis
- HTTP Method : POST
- Action : ListStreams
- Role : 前の手順で作成したIAMロールArnを入力します。

Integration Requestをクリックして、HTTP HeaderとMapping Templateを追加しておきます。

Mapping TemplateでContentタイプとしてapplication/jsonを追加して、空のテンプレートを作成しておきます。

Describe　Streamsのリソースとメソッド

プロキシでリソース[/streams/{stream-name}]を作成しておきます。

次の設定でGET Methodを作成しておきます。
- Integration type : AWS Service
- Region : us-east-1
- AWS Service : Kinesis
- HTTP Method : POST
- Action : DescribeStreams
- Role : 前の手順で作成したIAMロールArnを入力します。
HTTP HeaderとMapping Templateを追加しておきます。
Mapping TemplateでContentタイプとしてapplication/jsonを追加して、テンプレートに次のコードを追加しておきます。

Put Recordのリソースとメソッド

リソース[/streams/{stream-name}/record]を作成しておきます。

次の設定でPUT Methodを作成しておきます。
- Integration type : AWS Service
- Region : us-east-1
- AWS Service : Kinesis
- HTTP Method : POST
- Action : PutRecord
- Role : 前の手順で作成したIAMロールArnを入力します。
HTTP HeaderとMapping Templateを追加しておきます。
Mapping TemplateでContentタイプとしてapplication/jsonを追加して、テンプレートに次のコードを追加しておきます。

{
     "StreamName": "$input.params('stream-name')",
     "Data": "$util.base64Encode($input.json('$.Data'))Cg==",
     "PartitionKey": "$input.path('$.PartitionKey')"
}

APIのデプロイとテストする

Actionsで、Deploy APIを選択しておきます。
New Stageを選択して、ステージ名を入力して、Deployをクリックしておきます。

HTTP コマンドを使用してAPIをテストできます。次のコマンドを実行して、Kinesis Data Streamsにデータを送信します。

#Use HTTP PUT command to put data records in to Kinesis Streams
http put  API-Invoke-url/deploy-stage-name/list-stream-resource-name/kinesis-data-stream-name/data-to-be-sent

#Copy the Invoke Url from API Gateway console after deploying the api.
http put  https://id.execute-api.us-east-1.amazonaws.com/test-api/streams/api-data-s3/record Data='data to s3' PartitionKey=1