Kinesis FirehoseがS3に出力したファイルを１レコードずつ読み込む

AWS Python
2017年3月2日
鹿倉秀太

皆々様、お久しぶりのブログ更新です。

Kinesis FirehoseがS3に出力したファイルをPythonでモニャモニャしたいと思っていたところ以下のことでつまづいてしまいました。

１つのファイルに複数レコード出力されている場合、改行(区切り文字)がなく１レコードずつ読み込めない！

改行とか入っていて、１行ずつ読み込むんだろう・・・みたいな想像を勝手にしていたのでちょっとつまずいてしまいました。

AWSのデモデータをKinesisFirehoseに流すと実際のファイルはの中身は下記のようになります。

{"ticker_symbol":"TGH","sector":"FINANCIAL","change":-0.6,"price":65.43}{"ticker_symbol":"JYB","sector":"HEALTHCARE","change":-2.27,"price":42.95}{"ticker_symbol":"WAS","sector":"RETAIL","change":0.53,"price":13.04}{"ticker_symbol":"NGC","sector":"HEALTHCARE","change":0.32,"price":5.41}

1	{"ticker_symbol":"TGH","sector":"FINANCIAL","change":-0.6,"price":65.43}{"ticker_symbol":"JYB","sector":"HEALTHCARE","change":-2.27,"price":42.95}{"ticker_symbol":"WAS","sector":"RETAIL","change":0.53,"price":13.04}{"ticker_symbol":"NGC","sector":"HEALTHCARE","change":0.32,"price":5.41}

こういったJsonストリームデータをPythonで処理する時ってどうしたらいいんだろうと調べていたらjsonライブラリにJSONDecoderというのがあるのを見つけました。

下記サンプルです。

import json
import boto3

s3 = boto3.resource('s3')
bucket = s3.Bucket('backet name')
obj = bucket.Object('key name')
obj = obj.get()
json_stream = obj['Body'].read()

decoder = json.JSONDecoder()

while len(json_stream) > 0:

    record, index = decoder.raw_decode(json_stream)
    json_stream = json_stream[index:]
    print record

1

2

3

4

5

6

7

8

9

10

11

12

13

14

15

16

import json

import boto3

s3 = boto3.resource('s3')

bucket = s3.Bucket('backet name')

obj = bucket.Object('key name')

obj = obj.get()

json_stream = obj['Body'].read()

decoder = json.JSONDecoder()

while len(json_stream) > 0:

record, index = decoder.raw_decode(json_stream)

json_stream = json_stream[index:]

print record

これを実行すると下記のように出力されました〜。

{"ticker_symbol":"TGH","sector":"FINANCIAL","change":-0.6,"price":65.43}
{"ticker_symbol":"JYB","sector":"HEALTHCARE","change":-2.27,"price":42.95}
{"ticker_symbol":"WAS","sector":"RETAIL","change":0.53,"price":13.04}
{"ticker_symbol":"NGC","sector":"HEALTHCARE","change":0.32,"price":5.41}

1

2

3

4

{"ticker_symbol":"TGH","sector":"FINANCIAL","change":-0.6,"price":65.43}

{"ticker_symbol":"JYB","sector":"HEALTHCARE","change":-2.27,"price":42.95}

{"ticker_symbol":"WAS","sector":"RETAIL","change":0.53,"price":13.04}

{"ticker_symbol":"NGC","sector":"HEALTHCARE","change":0.32,"price":5.41}

そうです、この記事・・・単にストリームデータの読み込み方についての記事です！KinesisFirehoseは完全に引きです！

でもストリームデータを扱うことは個人的には今後も増えそうなので今回調べておいてよかったです！

ではでは〜。

ハンズラボ初の新卒新人研修やってます講師の視点から編

　臼井です。
最近は、4月に入社した、ハンズラボ初の新卒エンジニアのみなさま4名とキャッキャウフフする毎日です。
そう、ハンズラボも他企業様と同様に、絶賛新人研修中でなのです。

　写真は弊社ではなく、本日 One Day Outing にてお邪魔した、 Studio Geeks 様での新人さんの風景です。

Google 関連技術で著名な株式会社トップゲート様運営なので、ドロイドくんがいたり、GCP, Android 関連の書籍がいっぱいあります。

　突然ですが、説明のため、ハンズラボ(及び東急ハンズIT部門)における、
新人(新卒入社、もしくは東急ハンズ他部門からのIT部門・ハンズラボへの異動) メンバーの受け入れに関する現状を、以下に示します。

続きを読むハンズラボ初の新卒新人研修やってます講師の視点から編

みんなのPython勉強会#11に参加しました

Python 勉強会
2016年4月15日
MishiroSonoo

4月に新卒で入社した、美代と申します。
現在、新人研修でコンピュータの基礎を学習中です。

4月12日に開催された、「みんなのPython勉強会#11」に初めて参加したので、その内容と感想を簡単に書こうと思います。続きを読むみんなのPython勉強会#11に参加しました

はじめまして、ハンズラボにJOINした吉田です

AWS Slack Python Lambda
2016年2月7日
YoshidaYuki

皆様、はじめまして。
先日（1月中旬）ハンズラボにエンジニアとしてJOINした吉田です！
エンジニアブログ、頑張っていきます！

前職

2014年に新卒入社した会社でインフラエンジニアとして働いていました。
主にオンプレのWindowsServer（AD周り）の保守運用などの仕事に従事しつつ、AWSに恋をして夜な夜な一人でEC2と遊んでいました。
憧れのAWS環境でのお仕事なので頑張りつつ、認定資格5冠Tシャツ目指します。

1本目のエンジニアブログはLambdaに関して書いていこうと思います。

続きを読むはじめまして、ハンズラボにJOINした吉田です

AWS ElasticSearch Service の認証にIAM Roleを使う [Python編]

AWS ElasticSearch Service の認証にIAM Roleを使う [PHP編] の続き、
Python編です。

Python でどうするのか、検索してみたところ、github の elasticsearch-py のライブラリレポジトリにて mathom がAWS の認証機能を追加するパッチつくったよー的な議論があって、最終的にマージされたようです。
なので、普通に elasticsearch-py でちょっと設定すれば使えるようです。

AWS ElasticSearch service · Issue #280 · elastic/elasticsearch-py

elasticsearch-py は、ElasticSearchのOfficial low-level clientとのことです。
今見ると、elasticsearch-py のドキュメントに、こんな感じで使えるよって書いてありました。

Python Elasticsearch Client — Elasticsearch 2.2.0 documentation

from elasticsearch import Elasticsearch, RequestsHttpConnection
from requests_aws4auth import AWS4Auth

host = 'YOURHOST.us-east-1.es.amazonaws.com'
awsauth = AWS4Auth(YOUR_ACCESS_KEY, YOUR_SECRET_KEY, REGION, 'es')

es = Elasticsearch(
    hosts=[{'host': host, 'port': 443}],
    http_auth=awsauth,
    use_ssl=True,
    verify_certs=True,
    connection_class=RequestsHttpConnection
)
print(es.info())

1

2

3

4

5

6

7

8

9

10

11

12

13

14

15

from elasticsearch import Elasticsearch, RequestsHttpConnection

from requests_aws4auth import AWS4Auth

host = 'YOURHOST.us-east-1.es.amazonaws.com'

awsauth = AWS4Auth(YOUR_ACCESS_KEY, YOUR_SECRET_KEY, REGION, 'es')

es = Elasticsearch(

hosts=[{'host': host, 'port': 443}],

http_auth=awsauth,

use_ssl=True,

verify_certs=True,

connection_class=RequestsHttpConnection

)

print(es.info())

変更点も少なくて、ライブラリもそのまま使えていいですね。

カテゴリー: Python

前職

1本目のエンジニアブログはLambdaに関して書いていこうと思います。