takaaki.info

以前DynamoDBをLocalで建てるようにしたことでDynamoDBを使ったサービスの開発やテストがローカルで完結できるようになって幸せをかみしめていたのですが、このたびSQSも使うことになり、さてどうしたものかと思っていたらElasticMQというSQS互換インターフェイスを持った実装が存在することを知って、使ってみました。なお自分はElasticってつくとなんでもAWSのサービスみたいに感じてますが、ElasticMQとElasticSearchはAWSのサービスではないですね。

ElasticMQはJavaで実装されてるので、ダウンロードしてきて実行すればいいだけなので超簡単。開発も今のところ継続的に行われており、ロングポーリングもver 0.7から対応しているようで素敵です。

java -jar elasticmq-server-0.8.0.jar

1	java -jar elasticmq-server-0.8.0.jar

Pythonでbotoを使ってキューを登録してみます。

import sys
from boto.sqs import regioninfo
from boto.sqs.message import Message

AWS_ACCESS_KEY='TEST'
AWS_SECRET_KEY='TEST'

queue_name = 'test_queue'
name = "localhost"
endpoint = "localhost"

fake_region = regioninfo.SQSRegionInfo(name=name, endpoint=endpoint)
conn = fake_region.connect(aws_access_key_id="TEST", aws_secret_access_key="TEST", port=9324, is_secure=False);

queue = conn.create_queue(queue_name)
queue.write(Message(body='QUEUE TEST'))

import sys

from boto.sqs import regioninfo

from boto.sqs.message import Message

AWS_ACCESS_KEY='TEST'

AWS_SECRET_KEY='TEST'

queue_name = 'test_queue'

name = "localhost"

endpoint = "localhost"

fake_region = regioninfo.SQSRegionInfo(name=name, endpoint=endpoint)

conn = fake_region.connect(aws_access_key_id="TEST", aws_secret_access_key="TEST", port=9324, is_secure=False);

queue = conn.create_queue(queue_name)

queue.write(Message(body='QUEUE TEST'))

アクセスキーとシークレットはダミーですので何を指定してもいいっぽいです。続いてキューを読み出します。

import sys
from boto.sqs.message import Message
from boto.sqs import regioninfo

AWS_ACCESS_KEY='TEST'
AWS_SECRET_KEY='TEST'

queue_name = 'test_queue'
name = "localhost"
endpoint = "localhost"

fake_region = regioninfo.SQSRegionInfo(name=name, endpoint=endpoint)
conn = fake_region.connect(aws_access_key_id="TEST", aws_secret_access_key="TEST", port=9324, is_secure=False);

queue = conn.create_queue(queue_name)
queue.set_attribute('ReceiveMessageWaitTimeSeconds', 20)

i=0
while 1:
    msgs = queue.get_messages(10)
    for msg in msgs:
        print msg.get_body()
        queue.delete_message(msg)

import sys

from boto.sqs.message import Message

from boto.sqs import regioninfo

AWS_ACCESS_KEY='TEST'

AWS_SECRET_KEY='TEST'

queue_name = 'test_queue'

name = "localhost"

endpoint = "localhost"

fake_region = regioninfo.SQSRegionInfo(name=name, endpoint=endpoint)

conn = fake_region.connect(aws_access_key_id="TEST", aws_secret_access_key="TEST", port=9324, is_secure=False);

queue = conn.create_queue(queue_name)

queue.set_attribute('ReceiveMessageWaitTimeSeconds', 20)

i=0

while 1:

msgs = queue.get_messages(10)

for msg in msgs:

print msg.get_body()

queue.delete_message(msg)

読み出しのループを回しておいてキューにデータを入れると、読み出されているのがわかります。

ロングポーリングが正しく動いていない気がするので、今後調査の方向で。

開発環境をローカルで構築すると、ネットワークのない環境でも開発ができるので非常に便利です。特に昨年は飛行機に乗る機会がとても多く、僕の乗っていた便はまだ機内インターネット接続サービスの恩恵は受けられない便ばかりだったので、飛行機の中でも開発を続けるためには、開発環境とドキュメントをローカルに入れておくのはとても重要でした。VagrantとDashにはとてもお世話になりました。

さて最近DynamoDBを使ってサービス作っています。AWSのサービスは非常に便利でよいですが、DynamoDBはElastiCacheやRDSみたいにローカルにMemcachedやRedis、MySQL、PostgreSQLなど入れておけば代用できるわけでもないので、どうすんだろ、と思いました。

もう1点DynamoDBの問題として、DynamoDBの課金体系は保存容量以外に毎秒何回読み込みと書き込みができるかに対してお金を支払いますが、無料枠は、合計書き込み5回/秒、読み込み10回/秒までです。でもDynamoDBではグローバルセカンダリインデックスを用意すると、それぞれにつき少なくとも書き込み、読み込み1ずつ割り当てなければならないので、1つのテーブルが1回/秒で済まないケースもあって、本当にお試しでもない限り、無料枠をすぐ使い切っちゃいます。といっても、月数百円とかですむのですが、油断してて料金発生するのも怖い。

で、DynamoDBの互換のシステムをローカルで構築できる物があればいいのになって思って調べたら、Amazonが公式にDynamoDB Localなるものを公開していて、これを使ったらものすごく簡単にそれができてしまいました。

紹介記事は2013年9月ですがリンクは最新版を指すようになっていてダウンロードされたのは2014年1月8日版でした。2013年12月に発表されたグローバルセカンダリインデックスにも対応。Javaで書かれているのでjarファイルを実行すればサービスが起動してしまう。ポート番号はデフォルトが8000ですが、8000なんてなにか他のものと容易に衝突しそうなんでそこはは変えました。

java -Djava.library.path=. -jar DynamoDBLocal.jar --port 8088

1	java -Djava.library.path=. -jar DynamoDBLocal.jar --port 8088

それで、次はアクセスをしてみるのですが、開発はPythonを使うことが多いのでbotoを使いました。botoってconnectionの際にhostとportを指定できるようになってるんですね。

from boto.dynamodb2.layer1 import DynamoDBConnection
from boto.dynamodb2.table import Table
from boto.dynamodb2.items import Item
from boto.dynamodb2.fields import HashKey, RangeKey, GlobalAllIndex
import boto.dynamodb2.types

import hashlib
import datetime

conn = DynamoDBConnection(
        aws_access_key_id='123456',
        aws_secret_access_key='123456',
        host='localhost',
        port=8088,
        is_secure=False)

table_name = 'testtable'
tables = conn.list_tables()
if table_name not in tables['TableNames']:
    Table.create(table_name
        , schema=[
            HashKey('id', data_type=boto.dynamodb2.types.STRING),
            RangeKey('time', data_type=boto.dynamodb2.types.NUMBER)
        ]
        , throughput={'read': 1, 'write': 1}
        , global_indexes=[
            GlobalAllIndex('groupIndex',
                           parts=[HashKey('grp',data_type=boto.dynamodb2.types.STRING),
                                  RangeKey('time', data_type=boto.dynamodb2.types.NUMBER)],
                           throughput={'read': 1, 'write': 1})]
        , connection=conn
    )

time = int(datetime.datetime.now().strftime('%s'))
m = hashlib.md5()
m.update(str(time))
id = m.hexdigest()

table = Table(table_name, connection=conn)
item = Item(table, data={
    'id'  : id,
    'time': time,
    'grp' : id+"-foobar"
})
item.save()

results = table.query(grp__eq=id+"-foobar", index='groupIndex')
for res in results:
    print res['id'], res['time']

from boto.dynamodb2.layer1 import DynamoDBConnection

from boto.dynamodb2.table import Table

from boto.dynamodb2.items import Item

from boto.dynamodb2.fields import HashKey, RangeKey, GlobalAllIndex

import boto.dynamodb2.types

import hashlib

import datetime

conn = DynamoDBConnection(

aws_access_key_id='123456',

aws_secret_access_key='123456',

host='localhost',

port=8088,

is_secure=False)

table_name = 'testtable'

tables = conn.list_tables()

if table_name not in tables['TableNames']:

Table.create(table_name

, schema=[

HashKey('id', data_type=boto.dynamodb2.types.STRING),

RangeKey('time', data_type=boto.dynamodb2.types.NUMBER)

]

, throughput={'read': 1, 'write': 1}

, global_indexes=[

GlobalAllIndex('groupIndex',

parts=[HashKey('grp',data_type=boto.dynamodb2.types.STRING),

RangeKey('time', data_type=boto.dynamodb2.types.NUMBER)],

throughput={'read': 1, 'write': 1})]

, connection=conn

)

time = int(datetime.datetime.now().strftime('%s'))

m = hashlib.md5()

m.update(str(time))

id = m.hexdigest()

table = Table(table_name, connection=conn)

item = Item(table, data={

'id' : id,

'time': time,

'grp' : id+"-foobar"

})

item.save()

results = table.query(grp__eq=id+"-foobar", index='groupIndex')

for res in results:

print res['id'], res['time']

実際のDynamoDBのようにコンソールが用意されていてテーブル作成できるわけではありませんが、そもそもきちんとテーブル定義はコードとして残しておいたほうがいいですね。

AWSのブログ記事に以下の様なことが書かれています。

スループットの指定は無視される。スループット以上にアクセスすることはできてしまう。
アクセスキーとリージョンはDBファイル名を生成するときにだけ関係ある
シークレットキーは無視される
テスト用なので、プロダクション作らないほうがいいよ

ということで、これでとりあえずお金の心配も、ネットワークの心配もせずにDynamoDBで開発ができるようになりました。

takaaki.info

ソフトウエアデベロッパー兼技術系ライター水野貴明のサイト。開発・執筆関係のお仕事依頼お待ちしております。

Category Archives: aws

ElasticMQでAmazon SQSのローカルテスト環境を作る

DynamoDB Localでローカル開発環境構築