:>/dev/null

ラガードエンジニアの不撓不屈の精神/unlearning/go beyond

aws cuiを使用してKinesis streamデータ登録・確認

Kinesisはライブラリとして使用する事が多い為、実際の挙動を把握せず使ってるので調査した時の勉強メモ。
実際の挙動を把握する事で障害対応等で効率的に行動出来るようにする。

■ストリーム作成

aws kinesis create-stream --stream-name <ストリーム名> --shard-count <シャード数>

■レコード登録

# "ShardId": "shardId-000000000000" と出力されればOK
aws kinesis put-record --stream-name <ストリーム名> --partition-key 123 --data <レコードに保存したい文字列>

■レコード確認

aws kinesis get-shard-iterator --shard-id shardId-000000000000 --shard-iterator-type TRIM_HORIZON --stream-name <ストリーム名> --query 'ShardIterator'
{
    "ShardIterator": "AAAAAAAAAAE2R*******************=="
}

aws kinesis get-records --shard-iterator "AAAAAAAAAAE2R*******************=="
{
    "Records": [
        {
            "Data": "aG9nZWhvZ2U=",
            "PartitionKey": "123",
            "ApproximateArrivalTimestamp": 1546849899.19,
            "SequenceNumber": "49591791737946085028906975044094438871988285374419959810"
        },
        {
            "Data": "aG9nZTE=",
            "PartitionKey": "123",
            "ApproximateArrivalTimestamp": 1546851156.579,
            "SequenceNumber": "49591791737946085028906975051711880461380150184625831938"
        }
    ],
    "NextShardIterator": "AAAAAAAAAAFQkY****************==",
    "MillisBehindLatest": 0
}

■簡易確認の為、上記手順をスクリプト

$ cat chk-kinesis-record.sh
#!/bin/bash

KINESIS_NAME=<ストリーム名>

# 引数チェック
#echo "args = "$#
ARGS1=${1:?}
#echo $ARGS1

# レコード登録
aws kinesis put-record --stream-name $KINESIS_NAME --partition-key 123 --data $ARGS1

# shardIdからiterator値取得
SHARD_ITERATOR=$(aws kinesis get-shard-iterator --shard-id shardId-000000000000 --shard-iterator-type TRIM_HORIZON --stream-name $KINESIS_NAME --query 'ShardIterator')

# レコード取得
aws kinesis get-records --shard-iterator $SHARD_ITERATOR

kinesis→レコード登録・確認

$ ./chk-kinesis-record.sh hoge1  
{
    "ShardId": "shardId-000000000000",
    "SequenceNumber": "49591791737946085028906975051711880461380150184625831938"
}
{
    "Records": [
        {
            "Data": "aG9nZWhvZ2U=",
            "PartitionKey": "123",
            "ApproximateArrivalTimestamp": 1546849899.19,
            "SequenceNumber": "49591791737946085028906975044094438871988285374419959810"
        },
        {
            "Data": "aG9nZTE=",
            "PartitionKey": "123",
            "ApproximateArrivalTimestamp": 1546851156.579,
            "SequenceNumber": "49591791737946085028906975051711880461380150184625831938"
        }
    ],
    "NextShardIterator": "AAAAAAAAAAFQkY****************==",
    "MillisBehindLatest": 0
}

指定データをKinesisBase64 エンコーディング後、Dataに文字列が格納される。
Base64 デコーディングする事により元データ参照可能。
Base64 エンコーディング・デコーディングサイト: Base64 Decode and Encode - Online

また、時刻に関してはtimestamp形式で格納される。
UNIX時間(UNIX時刻)から日付に変換サイト: UNIX時間⇒日付変換 - 高精度計算サイト

■ストリーム削除

aws kinesis delete-stream --stream-name <ストリーム名>

AWS WEBコンソールからKinesisのData Streamの挙動モニタリング
AWS Kinesis → Data Stream → ストリーム名 → モニタリングへ移動

  • 下記図を確認
    • GetRecords.Bytes
    • GetRecords.Records
    • PutRecord.Bytes
    • PutRecords.Records

図の中に青い丸(単一データ)、又は青い線(複数データ)でグラフ化されているば正常

  • ※ 上部赤い線は制限値(Maxの値)なので無視で問題ない