Python3の対話シェル上でKinesis Streams上のレコードを読みだす

Pythonと `boto3` によるAmazon Kinesis Streamからのレコードの読みだしを、対話型シェル上でステップバイステップで実行してみました。現在のレコードをちょっと確認してみたいと思って確認した際のメモです。

# 前提

* APIアクセス用の認証情報がAWS CLIで[デフォルトプロファイル](https://docs.aws.amazon.com/ja_jp/cli/latest/userguide/cli-chap-configure.html#cli-quick-configuration)または[名前付きプロファイル](https://docs.aws.amazon.com/ja_jp/cli/latest/userguide/cli-chap-configure.html#cli-quick-configuration-multi-profiles)に格納されていること
* Pythonがインストールされていること（ここでは[Anaconda 2020.02 for Windows Installer](https://www.anaconda.com/distribution/)を使用しました）
* [boto3](https://boto3.amazonaws.com/v1/documentation/api/latest/guide/quickstart.html#installation)がインストールされていること

# 手順

1. [Boto3（AWS SDK for Python）](https://aws.amazon.com/jp/sdk-for-python/)のインポート
2. Kinesisのクライアントの生成
3. Streamの取得
4. StreamのShardの取得
5. StreamのShardのイテレータの取得
6. イテレータからのレコード取得

# 実行

[`boto3` をインポートしKinesisクライアントを生成](https://boto3.amazonaws.com/v1/documentation/api/latest/reference/services/kinesis.html#Kinesis.Client)します。デフォルトプロファイルを使うときは、2行目は `client = boto3.client("kinesis")` でいいはず。

~~~
>>> import boto3
>>> client = boto3.session.Session(profile_name="プロファイル名").client("kinesis")
~~~

レコードを取得する際にはイテレータが必要、イテレータを取得するためにはストリーム名、シャードID、等が必要、ということで順に取得していきます。

まず[`list_streams()`](https://boto3.amazonaws.com/v1/documentation/api/latest/reference/services/kinesis.html#Kinesis.Client.list_streams)でストリームの取得とストリーム名の確認。ストリーム名がリストで格納されているので、先頭のものを確認することにします。

~~~
>>> streams = client.list_streams()
>>> streams["StreamNames"][0]
~~~

次に[`list_shards()`](https://boto3.amazonaws.com/v1/documentation/api/latest/reference/services/kinesis.html#Kinesis.Client.list_shards)を使ってシャードの取得とシャードIDの確認。引数としてストリーム名を指定するので、先ほど取得したストリーム名の最初のものを使用します。シャードがリストで格納されているので、先頭のもののシャードIDを確認することにします。

~~~
>>> shards = client.list_shards(StreamName=streams["StreamNames"][0])
>>> shards["Shards"][0]["ShardId"]
~~~

これで[`get_shard_iterator()`](https://boto3.amazonaws.com/v1/documentation/api/latest/reference/services/kinesis.html#Kinesis.Client.get_shard_iterator)に必要な情報が集まりました。レコードを読み取るためのイテレータを取得します。`ShardIteratorType`については[公式ドキュメント（現時点で英語版のみ）](https://docs.aws.amazon.com/ja_jp/kinesis/latest/APIReference/API_GetShardIterator.html#Kinesis-GetShardIterator-request-ShardIteratorType)または[Developers.IOの記事](https://dev.classmethod.jp/cloud/kinesis-streams-at-timestamp-iterator/)などをご参照。

~~~
>>> iter = client.get_shard_iterator(StreamName=streams["StreamNames"][0], ShardId=shards["Shards"][0]["ShardId"], ShardIteratorType='TRIM_HORIZON')
~~~

これで[`get_records()`](https://boto3.amazonaws.com/v1/documentation/api/latest/reference/services/kinesis.html#Kinesis.Client.get_records)、このイテレータからレコードを読みだせます。レコードがリストで格納されているので、レコード数と先頭レコードの内容を確認することにします。

~~~
>>> records = client.get_records(ShardIterator=iter['ShardIterator'])
>>> len(records["Records"])
（件数が表示される）
>>> records["Records"][0]
（レコードの内容が表示される）
~~~

これでレコードを読みだすことができました。

# イテレータの有効期間

イテレータの有効期間は取得後5分間です。

> A shard iterator expires 5 minutes after it is returned to the requester.
> ([GetShardIterator - Amazon Kinesis Data Streams Service](https://docs.aws.amazon.com/ja_jp/kinesis/latest/APIReference/API_GetShardIterator.html#Kinesis-GetShardIterator-request-ShardIteratorType))

5分超が経ってから `get_records` で使おうとすると `botocore.errorfactory.ExpiredIteratorException` の例外が発生して失敗します。取得後はお早めにお使いください。

# まとめ

「Kinesisのレコードをちょっと確認してみたい」「スクリプト化する前にステップバイステップで`boto3`でのアクセス手順を確認してみたい」というタイミングがあったので、Pythonの対話型シェルでのKinesisのレコード読出し手順を確認、まとめてみました。上の説明では、ストリーム名やシャードIDのような手順上必要な情報だけを確認していますが、各段階でどんなデータが取得できているかを確認するのにも役立つと思います。

# 参考

主に以下を参考にしました。

* [AWS SDK for Python | AWS](https://aws.amazon.com/jp/sdk-for-python/)
* [Quickstart — Boto 3 Docs 1.12.20 documentation](https://boto3.amazonaws.com/v1/documentation/api/latest/guide/quickstart.html)
* [Kinesis — Boto 3 Docs 1.12.20 documentation](https://boto3.amazonaws.com/v1/documentation/api/latest/reference/services/kinesis.html)
* [GetShardIterator - Amazon Kinesis Data Streams Service](https://docs.aws.amazon.com/ja_jp/kinesis/latest/APIReference/API_GetShardIterator.html#Kinesis-GetShardIterator-request-ShardIteratorType)
* [[新機能]Kinesis Streamsが時刻ベースのイテレーターに対応しました ｜ Developers.IO](https://dev.classmethod.jp/cloud/kinesis-streams-at-timestamp-iterator/)

----
本ページ内容は筆者が参照の便のためにある時点でまとめた個人的なメモです。内容を保証するものではなく、また筆者の所属組織等とは一切かかわりがありません。