aws-cliからkinesisを使ってみた。

参考URL
https://docs.aws.amazon.com/ja_jp/streams/latest/dev/fundamental-stream.html

AWSのディベロッパー試験問題(udemy)をやっていると、やたらとkinesis data streams(大量のセンサやIoTからデータを取得・保持・応答するAWSサービス)の問題が出てくるので触ってみた。
kinesis専門用語
プロデューサ = データ生成するもの。センサ・IoT機器・PC・スマホ
シャード = プロデューサからデータを受け取り、コンシューマに引き渡す。デフォ２４時間保持。最大７日保持
※shard = 破片、ココではサーバの意味
コンシューマ = EC2,S3,DynamoDBがシャードにリクエストして、データをもらって何かする

windows10にaws cliをインストールして、DOS窓からaws操作してみた

1, 最初にkinesisインスタンスを生成。シャードは１個

aws kinesis create-stream --stream-name myKinesis --shard-count 1

# myKinesisの生成確認
aws kinesis describe-stream --stream-name myKinesis

aws kinesis create-stream --stream-name myKinesis --shard-count 1

# myKinesisの生成確認

aws kinesis describe-stream --stream-name myKinesis

2, プロデューサ役となって、データ送信。パーティーションキーは１固定(シャードが１個だから)。データはbase64エンコードする必要がある。
base64エンコードしていない文字列だと、Invalid base64: “test123″みたいに言われる。
成功すれば、シャードIDを返してくれる(データ取得時に使う)
SequenceNumberがデータIDみたいなもん？

aws kinesis put-record --stream-name myKinesis --partition-key 1 --data dGVzdDEyMw==
{
    "ShardId": "shardId-000000000000",
    "SequenceNumber": "49608216262475261301557993958278052112409396043306762242"
}

aws kinesis put-record --stream-name myKinesis --partition-key 1 --data dGVzdDEyMw==

{

"ShardId": "shardId-000000000000",

"SequenceNumber": "49608216262475261301557993958278052112409396043306762242"

}

3, コンシューマ役になって、データをもらうにはイテレータ取得→データ取得の２段階

# 1, シャードからイテレータ(反復子だけど、この場合はアドレスって意味？)を取得
aws kinesis get-shard-iterator --shard-id shardId-000000000000 --shard-iterator-type TRIM_HORIZON --stream-name myKinesis
{
    "ShardIterator": "AAAAAAAAAAFpgwTNeSs9llRlBDsXK9N7Yt55diaxLyNRZUeVWrXyxGVL9uTUJQjwbkfO6D1XzUigJINisEvCQXK5+3EEMGAmrztTX0Y3FCDIqUI2L22QKtPpql4RJsItBIt1bNGVzQILDG1DRAVCYoxGRERiep1Y4M4kttJN+Ef23YTQLvEWbn9qfQLYn1m6wOIOSrhWkaSpNNQlC7gXc6uLjIqt5noo"
}

# 2, そのイテレータから、データを取得(データはbase64エンコードされたまま)
# NextShardIteratorで、どんどんデータを取得していくのだろうか？
aws kinesis get-records --shard-iterator AAAAAAAAAAFpgwTNeSs9llRlBDsXK9N7Yt55diaxLyNRZUeVWrXyxGVL9uTUJQjwbkfO6D1XzUigJINisEvCQXK5+3EEMGAmrztTX0Y3FCDIqUI2L22QKtPpql4RJsItBIt1bNGVzQILDG1DRAVCYoxGRERiep1Y4M4kttJN+Ef23YTQLvEWbn9qfQLYn1m6wOIOSrhWkaSpNNQlC7gXc6uLjIqt5noo
{
    "Records": [
        {
            "SequenceNumber": "49608216262475261301557993958278052112409396043306762242",
            "ApproximateArrivalTimestamp": "2020-06-23T12:03:45.694000+09:00",
            "Data": "dGVzdDEyMw==",
            "PartitionKey": "1"
        }
    ],
    "NextShardIterator": "AAAAAAAAAAHTyf6A7zaXgHJxJx60sVoR+yxdHJuyAs2yLeTH/nHXHAtVT7i0rBMaSeFxo0f0DJ7PTGwaKvOjsDAgEDZIAElTkSFIruU9fJ4AWvBG3tnV1ynz9i5mZz8r+3RsdpxlGjjDVoCHs7KwfeJdbU0FGHeAzpdV8wuRrV6vdEoer4ZsOkJ54gID4xjXb0kCgKerCsbHJnaYNF5IcotDaUl8dhwm",
    "MillisBehindLatest": 0
}

# 1, シャードからイテレータ(反復子だけど、この場合はアドレスって意味？)を取得

aws kinesis get-shard-iterator --shard-id shardId-000000000000 --shard-iterator-type TRIM_HORIZON --stream-name myKinesis

{

"ShardIterator": "AAAAAAAAAAFpgwTNeSs9llRlBDsXK9N7Yt55diaxLyNRZUeVWrXyxGVL9uTUJQjwbkfO6D1XzUigJINisEvCQXK5+3EEMGAmrztTX0Y3FCDIqUI2L22QKtPpql4RJsItBIt1bNGVzQILDG1DRAVCYoxGRERiep1Y4M4kttJN+Ef23YTQLvEWbn9qfQLYn1m6wOIOSrhWkaSpNNQlC7gXc6uLjIqt5noo"

}

# 2, そのイテレータから、データを取得(データはbase64エンコードされたまま)

# NextShardIteratorで、どんどんデータを取得していくのだろうか？

aws kinesis get-records --shard-iterator AAAAAAAAAAFpgwTNeSs9llRlBDsXK9N7Yt55diaxLyNRZUeVWrXyxGVL9uTUJQjwbkfO6D1XzUigJINisEvCQXK5+3EEMGAmrztTX0Y3FCDIqUI2L22QKtPpql4RJsItBIt1bNGVzQILDG1DRAVCYoxGRERiep1Y4M4kttJN+Ef23YTQLvEWbn9qfQLYn1m6wOIOSrhWkaSpNNQlC7gXc6uLjIqt5noo

{

"Records": [

{

"SequenceNumber": "49608216262475261301557993958278052112409396043306762242",

"ApproximateArrivalTimestamp": "2020-06-23T12:03:45.694000+09:00",

"Data": "dGVzdDEyMw==",

"PartitionKey": "1"

}

"NextShardIterator": "AAAAAAAAAAHTyf6A7zaXgHJxJx60sVoR+yxdHJuyAs2yLeTH/nHXHAtVT7i0rBMaSeFxo0f0DJ7PTGwaKvOjsDAgEDZIAElTkSFIruU9fJ4AWvBG3tnV1ynz9i5mZz8r+3RsdpxlGjjDVoCHs7KwfeJdbU0FGHeAzpdV8wuRrV6vdEoer4ZsOkJ54gID4xjXb0kCgKerCsbHJnaYNF5IcotDaUl8dhwm",

"MillisBehindLatest": 0

}

4, 使わないkinesisを削除する
aws kinesis delete-stream –stream-name myKinesis

# 削除されたか確認
aws kinesis describe-stream –stream-name myKinesis
#　NotFoundになれば削除された。
An error occurred (ResourceNotFoundException) when calling the DescribeStream operation: Stream myKinesis under account xxxxxxxxxxx not found.

// パーティションキーと複数シャードの割当について
パーティションキーをハッシュ化(128bit)して、保持しているシャードに割り当ててるらしい。
128bitだと分かりづらいので2bitハッシュで考えてみると

パーティションキー=0ならmod4してシャード0に割り当て
パーティションキー=1ならmod4してシャード1に割り当て
パーティションキー=2ならmod4してシャード2に割り当て
パーティションキー=3ならmod4してシャード3に割り当て
パーティションキー=4ならmod4してシャード0に割り当て
パーティションキー=5ならmod4してシャード1に割り当て
パーティションキー=6ならmod4してシャード2に割り当て
パーティションキー=7ならmod4してシャード3に割り当て
パーティションキー=8ならmod4してシャード0に割り当て
パーティションキー=9ならmod4してシャード1に割り当て

みたいなイメージだと思う。

// 適切なシャード数は？
１シャード当たりの性能が、Write=1MB/sec、Read=2MB/secなので、センサから毎秒3MBとかデータ送信されているなら、最低３シャード必要。少し余裕をもつ必要がある。
シャード数の変更(リシャード)は出来るが、一度の操作では、２倍(分割)か1/2(結合)にしか出来ない。

まあ、なんとなくイメージは掴めてきた。実務ではあんまり使う機会がなさそう・・・。

関連記事