Kinesisストリームを作ってレコードを投げる【Node.js/TypeScript】


## はじめに
[aws-sdk/client-kinesis](https://www.npmjs.com/package/@aws-sdk/client-kinesis)を使えば、async/awaitで簡単にKinesisにデータを投げることができます。
試しに使うにあたり、ストリーム作成からやってみたので、それらの手順をまとめました。

未来の自分やどなたかのお役にたてれば嬉しいです。

### 環境
- OS: MacOS Catalina(10.15.7)
- Node: v14.16.1

### 注意
- Kinesisに無料枠はないので実行する度にお金がかかるらしい

## AWSでKinesisストリームを作成

- AWSにログインしKinesisサービスを検索
![image](https://user-images.githubusercontent.com/44778704/123534174-1c7c4e80-d756-11eb-84b5-7f07e26dab5c.png)
- 「データストリームを作成」をクリック
![image](https://user-images.githubusercontent.com/44778704/123534145-f3f45480-d755-11eb-9a1f-f13e30374505.png)
- 情報を入力
  - データストリーム名: test
  - 開いているシャードの数: 1
![image](https://user-images.githubusercontent.com/44778704/123534218-7f6de580-d756-11eb-9f1f-18342b31dd09.png)

- 「データストリームの作成」ボタンをクリック

参考: [AWS デベロッパーガイド](https://docs.aws.amazon.com/ja_jp/streams/latest/dev/tutorial-stock-data-kplkcl-create-stream.html)

## IAMの設定

### ポリシーの作成
- IAMサービスを検索しユーザ設定ページへ遷移
![image](https://user-images.githubusercontent.com/44778704/123534713-007aac00-d75a-11eb-935a-8784917c87cb.png)

- ポリシー設定画面でポリシーの作成をクリック
![image](https://user-images.githubusercontent.com/44778704/123534931-8fd48f00-d75b-11eb-80fd-84e7b9d97fe2.png)

- 対象サービスをKinesisに設定
![image](https://user-images.githubusercontent.com/44778704/123534977-e510a080-d75b-11eb-818b-c3f28a60fd12.png)

- アクションを以下に設定
  - DescribeStream
  - GetShardIterator
  - GetRecords
  - PutRecord

- リソースを先ほど作成したKinesisに設定(「このアカウント内のいずれか」にチェックを入れるとARNを選択できる)

- リクエスト条件を設定する
  - IPアドレスの場合は[確認くん](https://www.ugtop.com/spill.shtml)にアクセスし、「あなたのＩＰアドレス(IPv4)」に書いてあるIPアドレスをコピー

- ビジュアルエディタとJSONが以下のようになっていることを確認
![image](https://user-images.githubusercontent.com/44778704/123535308-f35fbc00-d75d-11eb-87b8-a1654290868c.png)
![image](https://user-images.githubusercontent.com/44778704/123535370-623d1500-d75e-11eb-8f81-43237320e5ed.png)

- 「次のステップ: タグ」をクリックする

- 何も入力せずに「次のステップ: 確認」をクリックする
- 内容を確認し「ポリシーの作成」をクリックする
![image](https://user-images.githubusercontent.com/44778704/123535512-39694f80-d75f-11eb-8b32-4a4f91f4b9bf.png)

### IAMにアタッチ

- ユーザ設定画面から「アクセス権限の追加」をクリック
- 先ほど作成したポリシーを検索して選択
![image](https://user-images.githubusercontent.com/44778704/123535580-b268a700-d75f-11eb-9cb6-03281beba81a.png)

参考: [Kinesis ポリシーの作成](https://docs.aws.amazon.com/ja_jp/streams/latest/dev/tutorial-stock-data-kplkcl-iam.html)

## Cledentialsの設定

- 右上の自分のアカウントメーをクリックし、「マイセキュリティ資格情報」を選択
![image](https://user-images.githubusercontent.com/44778704/123536979-42aaea00-d768-11eb-9a84-0c54b85a7264.png)

- ~/.aws/credentialsにアクセスキーとクレデンシャルを記入

```
 [default]
 aws_access_key_id = AXXXXXXXXXXXXXXXX
 aws_secret_access_key = XXXXXXXXXXXXXXXXXXXXXXXXXXXX

 [sample1]
 aws_access_key_id = AXXXXXXXXXXXXXXXX
 aws_secret_access_key = XXXXXXXXXXXXXXXXXXXXXXXXXXXX
```

## プロジェクトのセットアップ

```
mkdir kinesis
cd kinesis
npm init -y
npm i @aws-sdk/client-kinesis
```

## コードの記述

- app.jsを作成し以下を記載

```js

// セットアップ
const { KinesisClient, PutRecordCommand } = require('@aws-sdk/client-kinesis');
const REGION = 'ap-northeast-1';
const client = new KinesisClient({ region: REGION });

// 引数の定義
const params = {
  Data: Buffer.from(JSON.stringify({ id: 1, name: 'test' })),
  PartitionKey: 'partitionKey',
  StreamName: 'test',
};

// PutRecordコマンドの実行
const command = new PutRecordCommand(params);
const send = async () => {
  try {
    const data = await client.send(command);
    console.log(data);
  } catch (error) {
    console.log(error);
  } finally {
    console.log('done');
  }
};
send();
```


## 実行
- ルートディレクトリで以下を実行

```
node app.js
```

- 実行結果

```
{
  '$metadata': {
    httpStatusCode: 200,
    requestId: 'XXXXX-XXXXX-XXXXX-XXXXXX-XXXXXXXXXXXX',
    extendedRequestId: 'xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx/xxxxxxx/xxxxxxxxxxxxxxxxxxx/xxxxxxxxxxxx',
    cfId: undefined,
    attempts: 1,
    totalRetryDelay: 0
  },
  EncryptionType: undefined,
  SequenceNumber: '0000000000000000000000000000000000000000000000000000000000001',
  ShardId: 'shardId-000000000000'
}
```

httpStatusCodeが200なので成功している模様。

## 解説

### requireについて

```js
const { KinesisClient, PutRecordCommand } = require('@aws-sdk/client-kinesis');
```

`KinesisClient`は`KinesisClass`をnewするために必要なのでrequireしている。
`PutRecordCommand`はKinesisにレコードを1件追加する関数。

ちなみに他にもたくさんの関数があり、npm installしたソースコードの中身を覗いてみるとだいたいわかる。
よく使う&わかりやすいものだけピックアップすると以下のような感じ。

  - 場所:/node_modules/@aws-sdk/client-kinesis/dist/types/index.d.ts

```js
export * from "./KinesisClient";
export * from "./Kinesis";
export * from "./commands/CreateStreamCommand";
export * from "./commands/DecreaseStreamRetentionPeriodCommand";
export * from "./commands/DeleteStreamCommand";
export * from "./commands/DescribeStreamCommand";
export * from "./commands/GetRecordsCommand";
export * from "./commands/ListShardsCommand";
export * from "./commands/ListStreamConsumersCommand";
export * from "./commands/PutRecordCommand";
export * from "./commands/PutRecordsCommand";

// ..略。他にもたくさん書いてある

```
使う分だけrequireしておけばOK。


### params定義について

```js
const params = {
  Data: Buffer.from(JSON.stringify({ id: 1, name: 'test' })),
  PartitionKey: 'partitionKey',
  StreamName: 'test',
};
```

[AWS JavaScript SDK のドキュメント](https://docs.aws.amazon.com/AWSJavaScriptSDK/latest/AWS/Kinesis.html#putRecord-property)を参考に、引数を設定。今回は設定必須の値のみ渡している。このドキュメントによると、以下の値が設定できる。

``` js
var params = {
  Data: Buffer.from('...') || 'STRING_VALUE' /* Strings will be Base-64 encoded on your behalf */, /* required */
  PartitionKey: 'STRING_VALUE', /* required */
  StreamName: 'STRING_VALUE', /* required */
  ExplicitHashKey: 'STRING_VALUE',
  SequenceNumberForOrdering: 'STRING_VALUE'
};
```

- `Data`: kinesisに投げるデータをBase64にエンコードして設定する。
- `PartitionKey`: 複数のシャードに振り分ける際に振り分けルール。
の元になる値(?)。今回はシャード数が1なのでなんでもOKだと思う。公式には以下のように書かれている。

>パーティションキーは、ストリーム内のデータをシャード単位でグループ化するために使用されます。Kinesis Data Streams は、ストリームに属するデータレコードを複数のシャードに配分します。この際、各データレコードに関連付けられたパーティションキーを使用して、配分先のシャードを決定します。パーティションキーは Unicode 文字列で、各キーの最大長は 256 文字に制限されています。MD5 ハッシュ関数を使用して、パーティションキーを 128 ビットの整数値にマッピングし、関連付けられたデータレコードをシャードにマッピングします。後者のマッピングにはシャードのハッシュキー範囲を使用します。アプリケーションは、ストリームにデータを配置するときに、パーティションキーを指定する必要があります。

参考:[ -Kinesis デベロッパーガイド-](https://docs.aws.amazon.com/ja_jp/streams/latest/dev/key-concepts.html#partition-key)

- `StreamName`: ストリームの名前。



## 補足1 credentialsについて
認証情報の設定を忘れていた。
共有認証情報ファイルを~/.aws/credentialsに置いているので勝手に動いたが、本来は明示的に設定したほうがよさそう。
以下のようにKinesisClientをnewする時にcredentialsを渡せばOK。

```js
const aws = require('aws-sdk')
const credential = new aws.SharedIniFileCredentials()
new KinesisClient({ region: 'ap-northeast-1', credentials: credential })
```

## 補足2 TypeScript

最終的にTypeScriptを使った。

```
npm i typescript
npm i --save-dev @types/node
```

package.jsonに以下が追加される。
- "typescript": "^4.3.5"
- "@types/node": "^16.0.0"

コードはだいたい以下のように書いている。

- app.ts

```js
import { KinesisClient, PutRecordCommand, PutRecordCommandOutput, PutRecordInput } from '@aws-sdk/client-kinesis'

const client = new KinesisClient({ region: 'ap-northeast-1'})

const params: PutRecordInput = {
  Data: Buffer.from(JSON.stringify({ id: 1, name: 'test' })),
  PartitionKey: 'partitionKey',
  StreamName: 'test',
}

const command = new PutRecordCommand(params);
const send = async () => {
  try {
    const data:PutRecordCommandOutput = await client.send(command);
    console.log(data);
  } catch (error) {
    console.log(error);
  } finally {
    console.log('done');
  }
};
send();

```

使用する(使用しそうな)値と型はこんなかんじ。

- `REGION`: string
- `STREAM_NAME`: string
- `credential`: aws.SharedIniFileCredentials
- `client`: KinesisClient
- `params`: PutRecordInput
- `data`: PutRecordCommandOutput

## まとめ

公式のドキュメントはシンプルで非常にわかりやすいのですが、

- paramsがわからない
- 使える関数がわからない
- TypeScriptの型がわからない

などのちょっとしたつまずきがありました。
次にKinesisを操作するときは、この記事を読み返せばスムーズに書けるんじゃないかな、と思います。

以上