【AWS】Amazon Kinesis(Kinesis Data Streams)について解説

最近大量のデータを取り扱う案件があり、Amazon Kinesis Data Streams について学んだことをまとめて解説します。

【AWS】Amazon Kinesis(Kinesis Data Streams)について解説

 

 

Amazon Kinesis とは

Amazon Kinesis はストリーミングデータ(ストリームデータ)をリアルタイムで収集・保存・処理・分析をするサービスです。

 

Amazon Kinesis には以下のサービスがあります。

 

データの処理はリアルタイムなので、一旦 Oracle などのデータベースに保存して、そこから取り出すといったことはしません。

 

■ストリーミングデータ(ストリームデータ)とは

簡単に言いますと、終わりがなく継続的に生成されるデータを言います。

通常データの送信元は多数を想定しています。

 

例えば以下のデータが考えられます。

 

 

 

AWS Kinesis の利用例

AWS IoT と組み合わせて利用するケースが多いです。

 

 

 

Kinesis の料金

Amazon Kinesis Data Streams の料金

https://aws.amazon.com/jp/kinesis/data-streams/pricing/

 

Kinesis は従量課金制です。

そのため、使用したリソース分のみ料金を支払います。

 

以下の4つに対して費用が発生します。

 

Amazon Kinesis の料金例(東京リージョンの場合)

2019年1月時点での東京リージョンの場合の Amazon Kinesis の料金例です。

 

■シャード時間

1シャードの価格は 0.0195 USD/時間

1 USD ← 110円(2019年1月)

0.0195×110円=2.145円/時

2.145円×24時間=51.48円/日

51.48円×30日=1,544.4円/月

 

 

■PUT ペイロードユニット

1,000,000 ユニットごと $0.0215

0.0215×110円=2.365円

 

 

■データ保持期間延長(最長7日観)

シャード時間ごと $0.0260

0.0215×110円=2.365円

 

 

■拡張ファンアウト

 

 

 

 

AWS CLI(aws コマンド)から Amazon Kinesis Data Streams を使ってみる

以下の記事を参考にして aws コマンドより Amazon Kinesis Data Streams を使ってみてどのようなものか調べてみました。

有益な情報ありがとうございます。

 

AWS CLIを使って初めてAmazon Kinesisを使ってみた

https://qiita.com/toshihirock/items/86a7d5144ad54b18bf36

 

 

AWS CLI(aws コマンド)のインストール手順は以下を参考にしてください。

【AWS】EC2 インスタンスに AWS CLI(aws コマンド)をインストールする手順【Windows&Linux】

 

 

環境の設定

最初に環境を確認及び設定します。

 

■最初の状態

最初は何もない状態です。

【AWS】Amazon Kinesis(Kinesis Data Streams)について解説

 

環境の確認です。

OS は RedHat 7.6 です。

[root@redhat7 ~]# cat /etc/redhat-release
Red Hat Enterprise Linux Server release 7.6 (Maipo)
[root@redhat7 ~]#

 

 

aws コマンドのバージョンです。

[root@redhat7 ~]# aws --version
aws-cli/1.16.92 Python/2.7.5 Linux/3.10.0-957.el7.x86_64 botocore/1.12.82
[root@redhat7 ~]#

 

 

Kinesis 検証用にユーザーを作成します。

【AWS】Amazon Kinesis(Kinesis Data Streams)について解説

 

 

 

 

 

「ユーザー詳細の設定」でユーザー名を設定します。

AWS アクセスの種類は「プログラムによるアクセス」にチェックを入れて「次のステップ:アクセス権限」ボタンをクリックします。

【AWS】Amazon Kinesis(Kinesis Data Streams)について解説

 

 

 

 

「アクセス許可の設定」「既存のポリシーを直接アタッチ」をクリックします。

ポリシーは今回の場合は「AdministratorAccess」を選択して「次のステップ:タグ」ボタンをクリックします。

【AWS】Amazon Kinesis(Kinesis Data Streams)について解説

 

 

 

 

「タグの追加(オプション)」画面で必要な場合はタグを入力し「次のステップ:確認」ボタンをクリックします。

【AWS】Amazon Kinesis(Kinesis Data Streams)について解説

 

 

 

最後に設定内容を確認し「ユーザーの作成」ボタンをクリックします。

【AWS】Amazon Kinesis(Kinesis Data Streams)について解説

 

 

 

 

 

下図のように「成功」画面が表示されたら「.csv のダウンロード」ボタンをクリックします。

【AWS】Amazon Kinesis(Kinesis Data Streams)について解説

 

 

 

CSV の中身は下図のようになっています。

【AWS】Amazon Kinesis(Kinesis Data Streams)について解説

 

 

 

次に aws configure コマンドで環境の設定をします。

[root@redhat7 ~]# aws configure 
AWS Access Key ID [None]: AKXXXXXXXXXXXXXXXXX 
AWS Secret Access Key [None]: tRXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXX 
Default region name [None]: ap-northeast-1
Default output format [None]: json 
[root@redhat7 ~]#

 

 

 

 

Kinesis Data Streams の作成

次に Kinesis Data Streams を作成します。

 

kinesis 関連のコマンド一覧は以下のコマンドで確認できます。

[root@redhat7 ~]# aws kinesis help

 

KINESIS()                                                            KINESIS() 
 
 
 
NAME 
       kinesis - 
 
DESCRIPTION 
       Amazon  Kinesis  Data  Streams is a managed service that scales elastically for real-time processing of streaming big data. 
 
AVAILABLE COMMANDS 
       o add-tags-to-stream 
       o create-stream 
       o decrease-stream-retention-period 
       o delete-stream 
       o deregister-stream-consumer 
       o describe-limits 
       o describe-stream 
       o describe-stream-consumer 
       o describe-stream-summary 
       o disable-enhanced-monitoring 
       o enable-enhanced-monitoring 
       o get-records 
       o get-shard-iterator 
       o help 
       o increase-stream-retention-period 
       o list-shards 
       o list-stream-consumers 
       o list-streams 
       o list-tags-for-stream 
       o merge-shards 
       o put-record 
       o put-records 
       o register-stream-consumer 
       o remove-tags-from-stream 
       o split-shard 
       o start-stream-encryption 
       o stop-stream-encryption 
       o update-shard-count 
       o wait 
 
                                                                     KINESIS() 
(END)

 

 

create-stream でストリームを新規作成します。

[root@redhat7 ~]# aws kinesis create-stream --stream-name Test-Kinesis-01 --shard-count 1
[root@redhat7 ~]#

 

 

コマンド実行後に再度 Kinesis のダッシュボードにアクセスをすると Kinesis データストリームが作成されています。

【AWS】Amazon Kinesis(Kinesis Data Streams)について解説

 

 

設定内容を確認します。

[root@redhat7 ~]# aws kinesis describe-stream --stream-name Test-Kinesis-01 

    "StreamDescription": { 
        "KeyId": null, 
        "EncryptionType": "NONE", 
        "StreamStatus": "ACTIVE", 
        "StreamName": "Test-Kinesis-01", 
        "Shards": [ 
            { 
                "ShardId": "shardId-000000000000", 
                "HashKeyRange": { 
                    "EndingHashKey": "340282366XXXXXXXXXXXXXXXXXX", 
                    "StartingHashKey": "0" 
                }, 
                "SequenceNumberRange": { 
                    "StartingSequenceNumber": "495921900XXXXXXXXXXXXXXXXXXXX783334633474" 
                } 
            } 
        ], 
        "StreamARN": "arn:aws:kinesis:ap-northeast-1:XXXXXXXXXXXX:stream/Test-Kinesis-01", 
        "EnhancedMonitoring": [ 
            { 
                "ShardLevelMetrics": [] 
            } 
        ], 
        "StreamCreationTimestamp": 1547966148.0, 
        "RetentionPeriodHours": 24 
    } 

[root@redhat7 ~]#

 

 

 

jq コマンドでステータスを抽出します。

[root@redhat7 ~]# aws kinesis describe-stream --stream-name Test-Kinesis-01 | jq '.StreamDescription.StreamStatus'
"ACTIVE"
[root@redhat7 ~]#

 

 

 

list-streams コマンドで現在の設定を表示できます。

[root@redhat7 ~]# aws kinesis list-streams 

    "StreamNames": [ 
        "Test-Kinesis-01" 
    ] 

[root@redhat7 ~]#

 

 

 

Kinesis へのデータのプットとゲット手順

Kinesis にデータをプットします。

[root@redhat7 ~]# aws kinesis put-record --stream-name Test-Kinesis-01 --partition-key test123 --data Test-Data-123 

    "ShardId": "shardId-000000000000", 
    "SequenceNumber": "49592190084917990801368718240189236203776340937481912322" 

[root@redhat7 ~]#

 

 

 

最初にシャードイテレーターをゲットします。

[root@redhat7 ~]# aws kinesis get-shard-iterator --shard-id shardId-000000000000 --shard-iterator-type TRIM_HORIZON --stream-name Test-Kinesis-01 

    "ShardIterator": "AAAAAAAAAAFKPFBEYi6XOCK02mE187iScMSkhHRvUN+VFKBpiielGbfXnBJkdG0kC6vh5pF/O3wNUqJxq6wKq80Hra4I4OOFT1CoSvBxluEfivkHm67ICX6rInIXC5qMBUrQsPtJ2/Y7xUseRorbZbKUuIx3pqtTtZDiLlsV4Ty1i1ArrA4+2zJQLwI35Ubin6wa2qIjQh+8eRhpC9Hci47y/nZvAVq3" 

[root@redhat7 ~]#

 

 

 

get-record コマンドで先ほど取得したシャードイテレーターを指定してデータをゲットします。

[root@redhat7 ~]# aws kinesis get-records --shard-iterator AAAAAAAAAAFKPFBEYi6XOCK02mE187iScMSkhHRvUN+VFKBpiielGbfXnBJkdG0kC6vh5pF/O3wNUqJxq6wKq80Hra4I4OOFT1CoSvBxluEfivkHm67ICX6rInIXC5qMBUrQsPtJ2/Y7xUseRorbZbKUuIx3pqtTtZDiLlsV4Ty1i1ArrA4+2zJQLwI35Ubin6wa2qIjQh+8eRhpC9Hci47y/nZvAVq3 

    "Records": [ 
        { 
            "Data": "VGVzdC1EYXRhLTEyMw==", ← Base64 でエンコードされています。
            "PartitionKey": "test123", 
            "ApproximateArrivalTimestamp": 1547994158.081, 
            "SequenceNumber": "49592190084917990801368718240189236203776340937481912322" 
        } 
    ], 
    "NextShardIterator": "AAAAAAAAAAFYG9GwCdCuS6uDWTzsA0dA7qL8ERvCRAOJlPO6QRcIdwtnJvG5dKO79qu2Z6eBv0JC2g8uVX1R88XSvP4lBC7Q0m8jw32msunJ6DHYmeSwngYVX3r4MBpYpEOSp/l9+PMHHkmHuEi13sJ8+Zm/COjVqfO6cwRn/0RwOE83qIMWPoQS41W6fb3jAeX3h48n10ddppEFiHdGSgc3/fZ2JyGe", 
    "MillisBehindLatest": 0 

[root@redhat7 ~]#

 

 

base64 コマンドでデコードします。

「Test-Data-123」が表示されました。

[root@redhat7 ~]# echo "VGVzdC1EYXRhLTEyMw==" | base64 --decode
Test-Data-123[root@redhat7 ~]#

 

 

 

「Test-Data-123」をエンコードしてみます。

[root@redhat7 ~]# echo "Test-Data-123" | base64
VGVzdC1EYXRhLTEyMwo=
[root@redhat7 ~]#

 

 

 

 

参考サイト

AWS Black Belt Online Seminar 2017 Amazon Kinesis

https://www.slideshare.net/AmazonWebServicesJapan/aws-black-belt-online-seminar-2017-amazon-kinesis?qid=f35f6c6d-2622-44f5-b908-4d83e45117d5&v=&b=&from_search=9

 

Posted by 100%レンタルサーバーを使いこなすサイト管理人

コメントを残す

メールアドレスが公開されることはありません。 * が付いている欄は必須項目です

このサイトはスパムを低減するために Akismet を使っています。コメントデータの処理方法の詳細はこちらをご覧ください