Data Stream Processing and Analysis on AWS: Fluentd, Elasticsearch, DynamoDB, EMR and Amazon Kinesis

AWS Casual Talks #2 @suzu_v

すずけん, @suzu_v

VOYAGE GROUPで働いてます

その中のadingoという会社でR&Dをしてます

http://suzuken.hatenablog.jp/

アンケート

Amazon Kinesis ユーザの方?

Fluentd使っている方？

Elasticsearchユーザの方?

DynamoDB使ってる方?

EMR使ってる方?

今日はログの解析基盤のアーキテクチャと Kinesisの検証をしている話をします

Data Sources

App.4

[Machine Learning]

AWS

Endp

oint

App.1

[Aggregate & De-Duplicate]

Data Sources

Data Sources

Data Sources

App.2

[Metric Extraction]

S3

DynamoDB

Redshift

App.3 [Sliding

Window Analysis]

Data Sources

Availability Zone

Shard 1 Shard 2 Shard N

Availability Zone

Availability Zone

Amazon Kinesis Managed Service for Real-Time Processing of Big Data

http://www.slideshare.net/AmazonWebServices/amazon-kinesis-realtime-streaming-big-data-processing-applications-bdt311-aws-reinvent-2013

Kinesis Architecture

log

loglog

log

loglog

log

loglog

log

loglog

Amazon Kinesis

Kinesis Application

stream

Data Record

shard

Partition key

DataStore

Kinesis Architecture

log

loglog

log

loglog

log

loglog

log

loglog

Amazon Kinesis

Kinesis Application

stream

Data Record

shard

Partition key

DataStore

site_id2014/04/18 site_id: 1 user_id: 1

count by site 1~100

count by site 101~200

count by site 201~300 top ranking

Kinesis Architecture

log

loglog

log

loglog

log

loglog

log

loglog

Amazon Kinesis

Kinesis Application

stream

Data Record

shard

Partition key

DataStore

user_id2014/04/18 item_id: 1 user_id: 1

coefficient: item visited by user 1-10000

update matrix (in near real time)

coefficient: item visited by user 10001-20000

coefficient: item visited by user 20001-30000

Out System

システム概要

広告ログを分析するための基盤

アドホックな分析＆定常分析

ターゲティングにも使う

カジュアルなシステム要件

複数サービスのログをひたすら取り込む。

過去ログをひたすら取り込み、快適に分析できるようにする。hot / cold dataの分析を両立する。

ターゲティングはベストエフォート。

2012 mid.ELB

EC2

EC2 EC2

EC2

request

S3 MongoDB Cluster (EC2)

EMRELB

EC2

EC2 EC2

EC2

http api

crontab

hourly daily

php + apache

php + apache

Arch. #1 / 2012 mid.

Cons

not real-time

MongoDBのwriteによる負荷が高い

MongoDBの集計処理が重い

MapReduceジョブを回さないと分析ができない

Pros

MongoDBの柔軟性

データの受け入れが安定

2014 earlyELB

EC2

EC2 EC2

EC2

request

EC2

S3DynamoDB

EMR ELB

EC2

EC2 EC2

EC2

http api

Elastic search

Growth Forecast

fluentdfluentd

(aggregator)

exec_filter

out_elasticsearch

out_dynamodb

servlet (scala)

dynamic-dynamo

ある日のトラフィックパターンと自動スケーリングの様子

sebdah/dynamic-dynamodb https://github.com/sebdah/dynamic-dynamodb

Arch. #2 / 2014 early

Cons Fluentdが便利すぎるがために、aggregatorにいろいろやらせすぎてる。

Fluentdのバッファが詰まって死ぬ、とかある困る。

preノードを追加してストリームを逃がすこともしているが、fluentの設定が複雑化する

ストリーム処理をより柔軟に、多様に、疎結合に扱いたい

Elasticsearchに長期間の大量データを集計させるとそれなりにパワーが必要

Pros Elasticsearch + Kibanaにより、エンジニア以外のメンバーもhotなデータのドリルダウンを行えるように

DynamoDB導入によりwrite / readともに安定

dynamic-dynamodbを採用。throughputを自動調節するように。

EMR上のHiveから、s3にあるログとDynamoDBにある分析済みデータを組み合わせたクエリが可能に

準リアルタイムなターゲティングが可能に

ここから検証中の話

Next ?ELB

EC2

EC2 EC2

EC2

request

EC2

S3

DynamoDBEMR ELB

EC2

EC2 EC2

EC2

http api

Elastic search

Growth Forecast

Amazon Kinesis

Kinesis App.

Amazon Kinesisの検証項目

求めているスループットはでるか？

どれくらいの負荷で書き込めるか？

書き込み失敗時にどのような挙動になるか？ハンドリングできるか？

KCL(Kinesis Client Library)での開発は楽か？

ざっくりとした使用感 - producer

aggregatorを用意してwriteするのではなく、logの発生源から直接書くほうがthroughputを出しやすい。ただし、1レコードずつ書き込まなければならないため、fluentdのout pluginではthroughputが出しづらい。

1 shard 1000 put request / sec. の制限がある。大きなオブジェクトを書き込まない限り、byte制限よりこちらに引っかかるケースのほうが多い。とはいえ、シャードの追加は簡単なので、大きな問題にはならない。緩和申請は計画的に。

書き込み時の負荷を意識する必要がないので楽。とりあえず入れておけば、Consumer側でよしなにできる。

Kinesis内に24時間残ってくれるのは便利。ただし、Kinesisへの書き込み失敗時のハンドリングをする必要はある。

ざっくりとした使用感 - consumer

DynamoDBが活躍する。Kinesis ApplicationからDynamoDBへの書き込みは楽。Scalingも問題ない。

readは1 shard 5 transaction / sec.の制限がある。なので1transactionである程度まとめてデータを取得する作りにする必要がある。そもそも足りないならshardを増やす。

失敗時のretryのため、Kinesis Applicationの各Workerで各Shardをどこまで処理したのかをCheckpointとして記録しておく必要がある。KCL(Kinesis Client Library)使うとこのへんはよしなにDynamoDBで管理するようにしてくれる。

EMRからもConnectorがあるので簡単に扱える。

EMR上のSpark Streamingから扱うのは手軽で良い。EMRでSparkクラスタを立てられるので、aliveにしてストリームに対してwindowクエリを流し続けるのは手軽なユースケースになりそう。

Kinesisへの移行で発生する作業

fluentdのexec_filterとかexecとかでやっていたことをKinesis Applicationに置き換える必要がある

パーティションキーの考慮、consumeが失敗した場合のケア

fluentdでいうaggregatorノードを作るのではなくて、ログを出す各サーバから直接Kinesisに転送するような作りにしたほうが良さそう

1行のログごとの書き込みなので、1ノードからのKinesisへの書き込みスループットが現段階ではあまりでない。

Kinesis, こうなったら良い

Kinesisのメトリクスも1min.ごとに見たい

DynamoDB同様、batchWriteが欲しい

ap-northeast-1! 東京リージョン！

カジュアルなまとめ

Amazon Kinesisもっと使いましょう

東京リージョン待ってます

参考資料

Kinesis関連

Amazon Web Services ブログ: 【AWS発表】KinesisとElastic MapReduceを使ってストリーミングデータを処理する

http://aws.typepad.com/aws_japan/2014/03/process-streaming-data-with-kinesis-and-elastic-mapreduce.html

Amazon Kinesis: Real-time Streaming Big data Processing Applications (BDT311) | AWS re:Invent 2013

http://www.slideshare.net/AmazonWebServices/amazon-kinesis-realtime-streaming-big-data-processing-applications-bdt311-aws-reinvent-2013

https://www.youtube.com/watch?v=AXAaCG2QUkE

re:Invent 2013でのKinesisの紹介スライド及びビデオです。KeynoteでのKinesisのデモに利用されていたTwitterのhashtagをcountするデモの構築方法についてまとまっています。

ShardのマネジメントをDynamoDBを利用して行っているところなど、わかりやすく説明されています。

Kinesis関連

Developing Record Consumer Applications - Amazon Kinesis

http://docs.aws.amazon.com/kinesis/latest/dev/kinesis-record-processor-app.html

Kinesis Client Libraryを使ったConsumerの作成方法についてまとまっています

Tutorial: Analyzing Amazon Kinesis Streams with Amazon EMR and Hive - Amazon Elastic MapReduce

http://docs.aws.amazon.com/ElasticMapReduce/latest/DeveloperGuide/query-kinesis.html

Kinesis StreamをEMR上のHiveで取り扱う方法について説明されています。

Log4jAppender for Kinesis

Download Log4J Appender for Amazon Kinesis Sample Application, Sample Credentials File, and Sample Log File - Amazon Elastic MapReduce

http://docs.aws.amazon.com/ElasticMapReduce/latest/DeveloperGuide/kinesis-pig-publisher.html

てっとりばやくKinesisに書き込みたいならこれがいいです

Kinesis with EMR

Spark Streaming

http://www.cs.berkeley.edu/~marmbrus/sparkdocs/_site/streaming-programming-guide.html

cold dataとhot dataを組み合わせてwindowしたり

例）特定クライアントのキャンペーンにおける5分毎のクリック数

Run Spark and Shark on Amazon Elastic MapReduce : Articles & Tutorials : Amazon Web Services

https://aws.amazon.com/articles/Elastic-MapReduce/4926593393724923

credit

扉絵

https://www.flickr.com/photos/65220434@N02/13645528234/sizes/l

fluent-plugin-elasticsearch

uken/fluent-plugin-elasticsearch

https://github.com/uken/fluent-plugin-elasticsearch

hostsオプションおすすめ

Kibana向けにオプション作れるの便利

chunkサイズはES側の設定と合わせておくと良い

fluent-plugin-kinesis

imaifactory/fluent-plugin-kinesis

https://github.com/imaifactory/fluent-plugin-kinesis

まだrubygemsには上がってない

fluentのchunkと、Kinesisのwrite APIの相性が…

throughput出すならbatchで書き込めないとつらそう