Amazon Kinesis/Redshift編~アクセスログをkinesisで加工してTableauで表示してみよう② [全4回]~

こんにちは、中の人です。

Tableauは非常に強力なBIツールでエリア情報を地図にマッピングしてくれる機能などもあります。

IPアドレスのままではTableauに表示することが出来ないので、Amazon KinesisでIPアドレスを緯度経度に変換して、Redshiftに取り込んでTableauで表示させます。

全体構成

前回の『 Amazon Kinesis/Redshift編~アクセスログをkinesisで加工してTableauで表示してみよう~ 』に引き続き、今回は第2回としてKinesis Streamにputしたアクセスログを、Kinesis AppでgetしIPアドレスを緯度経度に変換してS3に保存します。

IPアドレスから緯度経度への変換を行うAPIは複数ありますが、検証した中では1番レスポンスが良かった以下を使用しております。

http://ip-json.rhcloud.com/

getrecordsToGEO.pyの作成

1$ vim getrecordsToGEO.py

※できるだけコメントで解説しています。

1#!/usr/bin/python23# -*- coding: utf-8 -*-45import boto.kinesis,datetime,time6import re7import json8import urllib29import os.path10import calendar11from urllib2 import Request, urlopen, URLError, HTTPError12from boto.s3.key import Key1314# kinesisで加工したログの保存先S31516# kinesis-demo-***はredshiftと同じリージョンにある任意のバケット名1718s3 = boto.connect_s3()19bucket = s3.get_bucket("kinesis-demo-***")20k = Key(bucket)2122dt = ""2324# stream_name(demo-knowcom)は自分のStream名に変更して下さい。2526connection = boto.kinesis.connect_to_region('us-east-1')27stream_name = 'demo-knowcom'28stream = connection.describe_stream(stream_name)29shards = stream['StreamDescription']['Shards'][0]['ShardId']3031kinesis_iterator = connection.get_shard_iterator(stream_name,shards,'LATEST')3233next_iterator = None34while True:35    if next_iterator is None:36        next_iterator = kinesis_iterator['ShardIterator']37    else:38        next_iterator = responce['NextShardIterator']3940    responce = None41    responce = connection.get_records(next_iterator,limit=1)4243    if len(responce['Records'])!=0:        44        result = responce['Records'][0]['Data']4546# 日付のフォーマット変換4748        resultArray = result.split(' [')49        dateTime = resultArray[1].split(']')[0]50        dateArray = dateTime.split('/')51        timeArray = dateArray[2].split(':')52        for i ,v in enumerate(calendar.month_abbr): 53            if dateArray[1]==v:54                dateMonth = i5556        eDatetime = timeArray[0]+'-'+str('%02d' % dateMonth)+'-'+dateArray[0]+' '+timeArray[1]+':'+timeArray[2]+':'+re.match(r'(.*) ',timeArray[3]).group(1)5758# その他のパラメータを取得5960        resultArray = result.split('"')61        eMethod = re.match(r'^([A-Z]*) ([^ ]*) (.*)', resultArray[1]).group(1)62        ePath = re.match(r'^([A-Z]*) ([^ ]*) (.*)', resultArray[1]).group(2)[:20]63        eProtocol = re.match(r'^([A-Z]*) ([^ ]*) (.*)', resultArray[1]).group(3)64        eStatus = re.match(r'^ ([0-9]*) ', resultArray[2]).group(1)65        eUserAgent = resultArray[5][:20]6667# elb配下のため、x-forwarded-forをIPとして取得6869        ipAddress = re.match(r'^ [0-9]* ([^ ]*) .*', resultArray[6]).group(1)7071# IPアドレスを緯度経度情報に変換7273        if re.search("^10\.",ipAddress):74            ipAddress = ""75        else:7677# ip ⇒ GEO変換用のAPI7879            url = 'http://ip-json.rhcloud.com/json/'+ipAddress80            req = urllib2.Request(url)81            try:82                fp = urlopen(req)83            except HTTPError, e:84                print 'Error code: ', e.code85            except URLError, e:86                print 'Reason: ', e.reason87            else:88                data = json.load(fp)89                od = dt90                d = datetime.datetime.today()9192# 分単位でログファイルを生成9394                dt = '%s%s%s%s%s' % (d.year, str('%02d' % d.month), str('%02d' % d.day), str('%02d' % d.hour), str('%02d' % d.minute))95                if od != dt:96                    logfile = 'access_log.' + od97                    if os.path.exists(logfile):98                            print logfile99                            k.key = logfile100                            k.set_contents_from_filename(logfile)101                print '%s, %s' % (str(data.get("latitude")), str(data.get("longitude")))102                if len(data)!=1:103                    fi = open('access_log.' + dt, 'a')104105                    fi.write(eDatetime + "\t" + eMethod + "\t" + ePath + "\t" + eProtocol + "\t" + eStatus + "\t" + eUserAgent + "\t" + ipAddress + "\t" + str(data.get("latitude")) + "\t" + str(data.get("longitude")) + "\n")106                    fi.close()107108# 1件づつ取得をしているため、Streamの帯域にあわせて制御109110    time.sleep(0.2)

では、早速アクセスログをkinesisにgetしてみましょう

1$ python getrecordsToGEO.py2337.4192008972, -122.057403564437.4192008972, -122.057403564537.4192008972, -122.057403564635.6850013733, 139.751403809737.4192008972, -122.057403564852.5, 5.75922.3500003815, 114.1333007811039.9289016724, 116.3882980351137.4192008972, -122.057403564

上記のようなログが出れば確認完了です。

プログラムのprint部分を変更することで、表示を止めることも可能です。

1-        print '%s, %s' % (str(data.get("latitude")), str(data.get("longitude")))
1$ python getrecordsToGEO.py &

でバックグラウンド実行させます。

次回はS3に保存したデータを1分おきにRedshiftにimportします。

お楽しみに!!

関連ソリューション

Amazon Kinesisを使ったデータ分析のご相談はこちら

この記事を書いた人

aws-recipe-user