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がシャードにリクエストして、データをもらって何かする
1, 最初にkinesisインスタンスを生成。シャードは１個
|
1 2 3 4 |
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みたいなもん？
|
1 2 3 4 5 |
aws kinesis put-record --stream-name myKinesis --partition-key 1 --data dGVzdDEyMw== { "ShardId": "shardId-000000000000", "SequenceNumber": "49608216262475261301557993958278052112409396043306762242" } |
3, コンシューマ役になって、データをもらうにはイテレータ取得→データ取得の２段階
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 |
# 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(結合)にしか出来ない。
まあ、なんとなくイメージは掴めてきた。実務ではあんまり使う機会がなさそう・・・。