こんにちは！ JQです。
前回は『 Kinesis編～Kinesisを試してみる①～ 』と題して、Kinesisを試してみました。
今回は『Kinesis編～Kinesisを試してみる②～』と題して、BotoからKinesisを試してみたいと思います。
環境準備
1.IAM Role付与インスタンスの起動
Botoから操作するためのインスタンスを起動します。
RoleにてKinesisの許可を与えておきます。
※詳細は以前の記事をご確認ください。
■Amazon EC2編～IAM roles for EC2インスタンスを立ちあげてみよう！～
スクリプト
2.スクリプトの作成
簡単な動作確認スクリプトを作成します。
下記では到着した新しいレコードからスタートします。
1getdata.py2#!/usr/bin/python34# -*- coding: utf-8 -*-5 6import boto.kinesis,datetime,time7 8connection = boto.kinesis.connect_to_region('us-east-1')9stream_name = 'StreamName’1011stream = connection.describe_stream(stream_name)12shards = stream['StreamDescription']['Shards'][0]['ShardId']13 14kinesis_iterator = connection.get_shard_iterator(stream_name,shards,'LATEST') 1516next_iterator = None17while True:18 if next_iterator is None:19 next_iterator = kinesis_iterator['ShardIterator']20 else:21 next_iterator = responce['NextShardIterator']2223 responce = None24 responce = connection.get_records(next_iterator,limit=1)25 print responce['Records']26 time.sleep(1)次にPutするスクリプトを作成します。
下記では日付をデータとしてPutします。
1#!/usr/bin/python23# -*- coding: utf-8 -*-45import boto.kinesis,datetime,time67connection = boto.kinesis.connect_to_region('us-east-1')8stream_name = 'StreamName’9partition_key = 'partition_1'10sampledata = datetime.datetime.now().strftime('%Y%m%d%H%M%S')1112put_data = connection.put_record(stream_name,sampledata,partition_key)3.確認
それでは実際に確認してみましょう！
Getを動かした状態でPutしてみます。
1# python getdata.py2[]3[]4[]5[]6[]78[{u'PartitionKey': u'partition_1', u'Data': '20140402164003', u'SequenceNumber': u'49538079870049146416514521285925208671557181958806044673'}]9[]Putしたデータを取得出来ました。
今回は動作確認の為に簡単なスクリプトで試していますが、
実際に利用する場合にはShardsやSequenceNumberの管理や、
アプリケーション側のロードバランシング、調整、エラーハンドリングやAuto Scalingが必要になります。
※Kinesisクライアント・ライブラリ(Java製)はその辺を簡素化してくれます。
いかがでしたでしょうか？
次回もお楽しみに！！！
