AWS Kinesis Firehoseを動かして見る #Python

はじめに

2015/10のAmazon re:Invent 2015でKinesis Firehoseが発表されました。

これみて、ビビッと来ました。(Client appやIOT機器で)Logを収集するServerを持っていない場合でも、溜めるだけならServerメンテ0でLogをS3やRedshiftに溜められるジャン！Google Analytics library見たいに、Beacon APIをCallすれば、何も考えずにRedshiftにLogが溜まって、Data解析超楽ジャン！これで、Amazonの掲げる2 tier Architectureに更に近づいたジャン！

と言う訳で、Firehoseの設定、Python scriptなど、調べた結果をチラシの裏しておきます。なお、この記事は、2015/11時点での調査結果なので、ご了承下さい。

結論

先に結論を書くと、Kinesis Firehoseの用途は、現時点では以下の様になっています。

Android Client Appや、Web Pageでの、使用/操作Logを取るためのBeaconっぽい用途には(まだ)使えない
Kinesis SDKのSupport 言語を見ると、Java, Node, Python, Ruby, AWS Cliとなっており、主にServer側のアプリへの埋め込みを想定して居る。NodeやRubyなどのServer側のLogを簡単にS3やRedshiftに放り込んでおける、と言うのがメインの用途な模様
合わせてkinesis agentという常駐serviceもReleaseしており、このfluentdっぽいServiceで、ServerのAccess/error Logなどを簡単にS3/Redshiftに放り込む事もできる

ちなみに、Java版AWS Firehose SDKをAndroid Appに組み込もうとしましたが、私のAndroid戦闘力が低すぎるのか、AWS Java SDKとAndroid Java Libraryの競合が解決出来ず、組み込む事は出来ませんでした。

と言う訳で、以下、FirehoseをPythonで動かして見ます。

Kinesis Firehoseの設定を行う

まず、受け取ったLogをS3に保存する、と言うCase想定で、Firehose側の設定を行います。AWS ConsoleのKinesisから、Go to Kinesis Firehoseを選択し、Create Delivery Streamを選択します。
(なお、Kinesis Firehoseは、現時点でUS Oregon等の一部Regionのみサポートです、Tokyoは動きません)

Step1 `Create Delivery Stream`

まず、受け取ったLogをどこに渡すか、の設定をします。

DestinationにAmazon S3に選びます
Delivery stream nameは適当な名前を(ここではKinesis-S3-Testとしています)
S3 bucketは、受け取ったLogを保存するS3 bucketを選択します(ここではkinesis-firehose-test1としています)
IAM　roleは、Kinesis→S3保存の為のroleです、Create - New Firehose Deliery IAM roleで専用のroleを作ります

Step2 `Configuration`

受け取ったLogをS3に書き出すTiming, 圧縮の有無などを設定します。Defaultのままでも良いですが、60秒毎にS3に書き出す設定にする為に、Buffer intervalを60に変えています(最小は60秒)。これで、Logが来ていれば、60秒毎にS3にfileとしてLogが書き出されます。

Step3 `Review`

最終確認です。特に問題なければ、Create Delivery Streamで設定完了です。

Python上でKinesis FirehoseにLogを送り付ける

つづいて、Python Scriptから作成したKinesis-S3-Testと言うDelivery Streamsに、logを送り付けます。

IAM Userの設定

Python scriptからKinesisにアクセスする為のIAM Userを作成します。AmazonKinesisFirehoseFullAccessのPermissionを付与して、API/Secret Keyを払い出します。

Python Scriptの設定

以下のPython codeを作成します。

accesskey, secretkeyは払い出したAPK/Secret Keyに変えて下さい(以下のCodeは適当なkeyです)
regionはここではUS Oregon(=us-west-2)としていますが、適宜変えて下さい
DeliveryStreamNameを先ほど作成した、Kinesis-S3-Testと設定して下さい
pip install boto3でboto3を入れて下さい。以前boto3を入れた人はpip install -U boto3でUpdateを一旦掛けて下さい

やっている事は、1秒毎にclient.put_recordでKinesis Firehoseに2015-11-13 13:34:21というformatの現時刻を送り付けています。


#!/usr/bin/env python
# -*- coding: utf-8 -*-

import boto3, datetime, time

def main():
  accesskey = 'AKIAABCDEFGHIJK2345PQ'
  secretkey = 'CUQVma+ilWkC7FOU8isueWKWUGk7GB'
  region    = 'us-west-2'
  client = boto3.client('firehose', aws_access_key_id=accesskey, aws_secret_access_key=secretkey, region_name=region)

  count = 0
  while count < 100 :
    response = client.put_record(
        DeliveryStreamName='Kinesis-S3-Test',
        Record={
            'Data': datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S')
        }
    )
    print(response)
    count += 1
    time.sleep(1)

if __name__ == '__main__':
  main()

S3 bucketの確認

AWS ConsoleのS3に行き、kinesis-firehose-test1というBucketを開きます。
kinesis-firehose-test1/YYYY/mm/DD/hhというフォルダの下に、Log fileが出来ていると思います。Python scriptで送ったlogが保存されているか確認して見て下さい。

まとめ

とりあえず、PythonからKinesis FirehoseにLogを送り付けてみました。kinesis agentと組み合わせてfluentdの置き換え用途には良いと思います。S3のBucketにLogを置いて、それをTriggerにLambda functionでAmazon Elasticsearchに登録をすれば、Elasticsearch/KibanaもAmazon管理でまかなえます。

ですが、Android, javascriptが今後サポートされれば、Beaconっぽく使えるはずなので、サポートおなしゃす。

AWS Kinesis Firehoseを動かして見る