AWS kinesis GetRecords の正しい呼び出し方

AWSのサービスの１つであるKinesis Streamを使う機会がありました。

APIを使ったデータのアップロードとダウンロードをする際にいくつか気を付けないといけない点があったので挙げておきたいと思います。

Kinesisの特徴

Kinesisの特徴はデータのリアルタイム処理ができることでクラウドにアップロードしたデータは基本的に２４時間の間アクセスできます。
RDBと違って永続的に保存されるわけではなく、あくまでストリームに流したリアルタイムなデータに対して何らかの解析を行うというのがサービスの用途のようです。

データをアップロードするAPI

さて、kinesisへのデータのアップロードは、

1. PutRecord
2. PutRecords

という2つのAPIによって行います。

両者の違いは文字通り１つのデータか複数データかになります。

ちなみに、PutRecordsはAPIの仕様を読む限り、複数レコードのアップロードのうち、一部のデータのみアップロードに成功し、一部は失敗するということがあるようです。

トランザクション的な機能が無く、またレスポンスの内容からはどこからどこまでが成功でどこからどこまでが失敗なのかが分からないようです。手堅くデータをあげるのであればPutRecordをつかえということなのかもしれません。

データの取得はどうするのか？

アップロードしたデータを取り出す時はどうすればいいかですが、

1. GetShardIterator

docs.aws.amazon.com

1. GetRecords

docs.aws.amazon.com

というAPIの呼び出しによりデータの取得が可能です。

さて、このGetShardIteratorとGetRecordsの使い方ですが、単純に

「こっからここまでのデータをくれ」

という感じでの呼び出しにはなりません。

特にGetRecordsの呼び出しは一般的なWebAPIにおけるデータ取得とは少しイメージが異なる形です。

細かい説明はAPIのリファレンスを見て頂ければ分かると思いますが、ここではRubyのaws-sdkを使ったPutRecordとGetRecordsを呼び出すような、サンプルプログラムを上げておきたいと思います。
特に、データを取得する処理に関して、GetRecordsを使われる方のご参考になれば幸いです。

require 'aws-sdk'

# AWS の設定用定数
REAGION = 'リージョン名'
ACCESS_KEY_ID = 'アクセスキー'
SECRET_ACCESS_KEY = 'シークレットアクセスキー'
STREAM_NAME = "ストリーム名"

Aws.config.update({
  region: REAGION,
  credentials: Aws::Credentials.new(ACCESS_KEY_ID, SECRET_ACCESS_KEY)
})

kinesis = Aws::Kinesis::Client.new(region: REAGION)

# PutRecordで上げるデータはBase64エンコードする必要がある
# →従って、文字列だけでなくバイナリデータもあげることができる
enc_data = ["123"].pack('m')

resp = kinesis.put_record({
  stream_name: STREAM_NAME,
  data: enc_data, # required
  partition_key: "key-test", # required
})

# shardIdにはputRecordのレスポンスに格納されているShardIdを設定する
# シャード数が1であれば恐らく "shardId-000000000000"になっている
shd_id = resp.shard_id
puts "shard_id:#{shd_id}"
puts "sequence_number:#{resp.sequence_number}"

resp = kinesis.get_shard_iterator({
  stream_name: STREAM_NAME, # required
  shard_id: shd_id,    　   # required
  shard_iterator_type: "TRIM_HORIZON"
}
)

shd_it = resp.shard_iterator

resp = kinesis.get_records({shard_iterator: shd_it, limit: 10000})

loop do
	resp = kinesis.get_records({shard_iterator: shd_it, limit: 10000})
  resp.records.each do |record|
	puts record
    end
    # millis_behind_latest が0になった場合に終了させる
    # get_records自体はデータがなくても成功する
    break if resp.millis_behind_latest == 0
    break if resp.next_shard_iterator.nil?
    shd_it = resp.next_shard_iterator
end

ポイント

GetRecordsの呼び出しで注意しないといけないポイントがあります。
GetRecordsのレスポンスとして返ってくるデータには必ずしもデータが入っているとは限りません。
NextShardIteratorがある限り呼び出しを行わないといけないのですが、実はNextShardIteratorだけでチェックしてもデータが全てとり終わったのかどうかが分かりません。

全てのデータがとり終わったかどうかは、MillisBehindLatestが0かどうかでチェックする必要があります。
このチェックがないとGetRecordsをいつまで呼び出していいか分からず無限ループにはまってしまいます。

感想

サンプルコードを書いてみたはいいのですが、私の中でkinesisやMQTTのようなサービスやプロトコルのメリットが今一つ理解できていない部分があります。

世の中には永続化すべきデータとそうでないデータがあり、両方合わせると莫大な量になるが、
「とりあえず集めるだけ集めて必要なデータだけを残そう」という思想なのかもしれません。

IoTが広がるにつれてkinesisのようなサービスはこれからどんどん使われるようになるでしょうか。。。

Amazon Web Services 基礎からのネットワーク&サーバー構築 改訂版

Amazon Web Services実践入門 (WEB+DB PRESS plus)

注意点

Kinesis Streamはデータの通信量による課金とは別に、作成したストリーム単位でも費用が発生するので、ほったらかしにしないよう注意してください。

4月の振り返りと読書録

4月が終わってGWに突入しました。

今年の自分の目標として、月に２本ブログに記事をあげることを１つのノルマとしているのですが、特に書くとがないので振り返りや読書録等を上げておきたいと思います。

書くことが無いというよりは、書きたい気分にならないというべきかもしれません。

この3月、4月は桜の景色を楽しむ暇もない忙しい春でした。

特に４月は子供達が小学校や幼稚園に入ることもあって私用も多く、また新しい仕事の引き合いも頂いて神経を使った１か月でした。

仕事の引き合いを頂いた場合、今取り組んでいる仕事の時間を削って新しい仕事に向けた時間を確保する必要があります。

お客様への満足度が下がらないように同時に物事を進めるというのはとても疲れることですし、結果としてうまくいかない場合もあります。

4月に忙しかったせいでなんだかどっと疲れたでたというかGWはのんびりと勉強したり普段できないことに時間を使いたいと思います。

最近読んだ本

書きたい気分ではないということもあるので最近読んだ本・読んでいる本を感想と一緒に上げておきたいと思います。

流通王・中内功とは何者だったのか

kindle本で読みました。
中内功さんは私の好きな経営者です。家系が土佐の方だと知ってすごく親近感がわきました。

ダイエーは結果として衰退しましたが中内さんの生き方・考え方にあこがれる部分があります。

マンガ　線形代数入門　はじめての人でも楽しく学べる (ブルーバックス)

ディープラーニングの勉強のために読んでみました。
私は文系出身なので数学は数ⅡBまでしか学んでいません。ディープラーニング・人工知能について勉強するためには行列の勉強からやるべきと思いよんでみました。
マンガということもあって分かりやすいのですが、この手の数学の入門書籍を何冊か読んでみて思ったことですが、数学はやはり自分で手を動かして計算してみてなんぼだということです。
自分で紙と鉛筆を用意して計算してみないと身につきませんね・・・

電話はなぜつながるのか 知っておきたいNTT電話、IP電話、携帯電話の基礎知識

kindle本で読みました。
CTI関係の仕事の話があったのですが、VoIP関係のプロトコルはもちろんの事、電話に関してはハードレベルのことなど何も分かっていません。
この本は電話や通信の基本的なレベルの内容から丁寧に説明されてとても分かりやすいです。
まだ読んでいる途中で、つまみ読みしかできていませんが、この本を一冊きちんと読めば電話に関する基礎知識はしっかりと身につくのではないかと思います。

振り返り

仕事が忙しくなればなるほど色々な本を読みたくなりついつい本屋さんやamazonで買ってしまうのですが読まずに終わる本も増えてしまいます。
GWは読書と勉強の時間をしっかり取って５月も頑張ろう。

根性の本質は何か

ちょっと気になる記事を最近読んで、「根性」という言葉について思ったことを上げておきたいと思います。

storys.jp

この記事の話の中で、上司の方が、

低学歴の奴は勉強する根性さえ身についていない怠け者や。そんな奴に仕事ができるわけがない。

とこの話の投稿者の方に語られますが、実はこの上司の方と私も同じようなことを最近感じていました。

去年の夏にシステムプログラミング会という集まりに参加させて頂いたのですが、参加されている方に東大を卒業されている方が多かったようでした。

mcommit.hatenadiary.com

３流プログラマの私からすると信じられないくらい優秀な方々で、話しかけるのも畏れ多いような感じもして、自分の無能さに腹が立つと同時に皆さんの才能をうらやましく思いました。

ところが、話を聞いていると(東大を卒業されるような優秀な方達でも)勉強を苦痛と感じることがあって、それでもそういう時間を乗り越えた上で現在にたどり着いている人もいるということでした。※もちろん才能・天才という言葉でしか説明がつかないような方もいらっしゃると思いますが。

やはり、

「努力って大切なんだな」

と考えさせられる経験になりました。

同時に、

自分の能力が低いということは、

• 努力していない
• やる気がない
• 根性が無い

といった側面があるのかもしれないと感じました。

根性というと体育会系のイメージがありますが、考えてみれば勉強するのも結構大変です。
地道に単語や年号を覚えたり、数学の難しい問題に取り組んだりするのはしんどかったりします。

甲子園に出場する野球部のイメージとかだと、部活の練習の時間は頑張って授業の時間は体力回復のために寝て勉強はまったくしないというイメージがあります。
※もちろん全ての高校球児がそうではないでしょうが、実際のところよくある話だと思います。

これって捉えようによっては、

好きなことだけやって、勉強はしないなんて根性無いやつや！

という見方のできると思います。

そもそも、

根性＝辛くても耐えられる力・乗り越えられる力

というイメージがありますが、要は好きなことや到達したい目標のためであれば人間頑張れるものなんだと思います。

傍からみたらなんでそんなに頑張っているのか分からない・・・
すごいな、俺には無理だな・・・

といった気持ちを表現するために根性という言葉があるんだと思いますが、当事者が、好きなことや目標のためにやっていることであればそれを根性という言葉で表現するのは適切ではないような気がします。

要するに、根性の本質は

取り組んでいることがどれくらい好きか？

ということだと思います。
そこまで好きでなければ途中でやめてしまうという感じでしょうか。

体育会系的根性のメリット

ちなみに、体育会系的な根性と勉強を通して身につく根性の違いはなんでしょうか。

思いつくのは他者との関わりです。

勉強は一人でするもので、自分との闘い、自分と向き合う時間になりますが、スポーツなどは他人との関わりがあります。(個人競技であっても部活等の練習は集団でする事になりますし)

そういう意味では、体育会系的根性のある人は人との関わり方に耐性があるという点で優れているのではないかと思います。

人間という言葉が人の間と書くように、社会人として、人と人との関わりの中でしっかりと生きていけるというのは

文武両道

結局のところ、

「体育会系的根性もあるし、勉強で培った根性もある」

といういわゆる文武両道型人間が最強の存在なのかもしれません。
※う～ん、なんか欠点が無い人間って嫌な気もしますが・・・

一番大切なことは何か？

記事を書いてみて思ったことですが、

「あいつは根性あるな」と

言われるくらい頑張れるもの・好きなものを見つけること

が何よりも大切だと思いました。

どうすれば c++ の eof に対する違和感を抑えることができるのか？

C++でバイナリファイルの読み込みをしたいとき、ソースコードはこんな感じか。

#include <stdio.h>
#include <iostream>
#include <fstream>

using namespace std;

int main(int argc, char **argv)
{
	string filePath = "test.bin";

	ifstream fin( filePath.c_str(), ios::in | ios::binary );
	if (fin) {
	    unsigned char c;

	    while(!fin.eof())
    	{
	    	fin.read( ( char * ) &c, sizeof(unsigned char));
    		printf("[0x%02X]", c);
	    }

	    fin.close();
	} else {
		return -1;
	}

    return 0;
}

↓読み込むデータ(上記コードのtest.bin)としてこんなデータを用意してみます。

出力

$ g++ main.cpp 

$ ./a.exe 

[0x00][0x01][0x02][0x03][0x04][0x05][0x06][0x07][0x08][0x09][0x0A][0x0B][0x0C][0x0D][0x0E][0xFF][0xFF]

おー！読めてる読めて、うん！？

[0xFF]が１個多い！？

何が起こってるんだ！？

調べたところ、eosはストリームの読み込みを行った上でファイルの終端に達したかを返してくれるそうです。
※上記コードはバイナリファイルの読み込みをしていますが、テキストファイルでもeosの挙動は同じです。

stackoverflow.com

stackoverflow.com

知らなかった・・・

C言語でもfgetcとかfgetsとか使うとループは終端まで読み込んでからEOFと比較するようなコードになります。

う～ん・・・C++ の eosの挙動もC言語の挙動との類似性を意識したものかもしれませんがどうもしっくりしません。

ビルドしたモジュールに使われているDWARFのバージョンを調べる

最近デバッガに興味を持ってDWARFのフォーマットについて調べていました。

DWARFの仕様については公式サイトで公開されています。

Download DWARF Standards

DWARFの仕様にはいくつかのバージョンが存在するのですが、

ビルド後のモジュールにどのバージョンのDWARF情報が含まれているのかがよく分かりませんでした。

いろいろ検索してみると、同じことを疑問に思われた方がいらっしゃいました。

stackoverflow.com

こちらのページに書かれている通り試してみると、

$ readelf --debug-dump=info a.out | grep -A 2 'Compilation Unit @' 

  Compilation Unit @ offset 0x0:                                                                                 

   Length:        0xba (32-bit)

   Version:       4

のような出力が得られて、確かに確認できました。

Compilation Unitにバージョンが格納されているんですね。