ソリューションアーキテクトの今井(@imai_factory)です。
今日はAWS Summit Tokyo 2014で東京リージョンのリリースが発表されたAmazon Kinesisと、これを便利に使うためのツールやライブラリのエコシステムを紹介したいと思います。これらは数も多いので何度かにわけてご紹介していきたいと思います。
初回となる今回はAmazon Kinesis自体についてのおさらいと、2014年8月時点でリリース済みのツール群のさわりを紹介します。ツールそれぞれのディープな話題については次回以降にご期待ください。
リアルタイムなデータ処理とAmazon Kinesis
このブログを読んでいる方のなかにはデータの可視化や集計について考えてみたり、実際に開発・運用している方が多いのではないかと思います。実装方法にはいろいろあると思いますが、AWS上で行うのであればAmazon S3とAmazon Elastic MapReduceやAmazon Redshiftを組み合わせて使うパターンが多いのではないかと思います。図にするとこんな感じでしょうか。
この方法の特徴としては15分や1時間、または1日に1回などのインターバルを定義し、計算対象ウィンドウ（前述の15分や1時間、1日のこと）のデータが完全に揃った状態で行われることが挙げられます。いわゆるバッチ処理ですね。インターバルを縮めていくことによって処理の計算粒度を上げていくことは可能ですが、あくまでスケジュールドリブンの処理方式です。
これに対する概念としてデータ入力によって駆動される処理方式がありますが、そのなかでも入力を逐次的に処理していく方式のことをストリームプロセッシングと呼び、入力されるデータをストリームと呼びます。Amazon Kinesisは流速と流量の大きなストリームを受け取り、後続のプロセスが処理しやすいよう、ユーザーが指定するPartition Keyでソートしてくれるサービスです。
たとえば先ほどのレポーティングのアーキテクチャに、Amazon Kinesisを組み合わせてリアルタイムダッシュボードを実装しようとすると以下のようなイメージになります。
もちろん、ストリームプロセッシングが実現するのはデータのリアルタイムな可視化だけではなく、他にも以下のようなユースケースが考えられます。
- ユーザーのWebサイト内での行動をトラッキングし、リアルタイムな（もしくはリアルタイムに近い）レコメンデーションを行う
- 工場等に施設に配置されたセンサー群からのデータを受け取り、例えば温度や湿度などのリアルタイムな異常値検知を行う
- 金融や広告のリアルタイム取引において、市場価格や落札価格などのスライディングウィンドウ分析（例えば移動平均）を行う
Amazon Kinesisがもたらすもの
ストリームプロセッシングでは、ストリームの上流でデータを生成するアプリケーションやデバイスのことをProducerと呼びます。例えば以下のようなものが挙げられます。
- アプリケーション・サーバー
- モバイルデバイス/モバイルアプリケーション
- ウェブブラウザ
- センサーデバイス
- カーナビ
- Wifiのアクセスポイント
一方、下流でデータを処理するアプリケーションのことをConsumerと呼ばれ、前述のユースケースで述べたユーザーやデバイスの状態トラッキングや集計、などを行います。
この間を取り持つのがAmazon Kinesisの役割であり、おおまかにいって以下の機能・ベネフィットを提供します。
- データのソート
- Kinesisへのデータ投入時にユーザーが指定したPartition Keyによって、後続のどのShardに出力されるかが決まります。Partition Keyはシャーディングのためのキーのようなものです。
- これにより、Consumerは自分のShardに流れてくるデータのみを気にしてあげればよいということになります。
- さらに、Data recordにはKinesisへ投入された順番にSequence numberが付与されます。これにより、Consumer側で「どこまで処理をしたのか」だったり「どこから処理を開始するのか」といった管理が容易になります。
- 柔軟なキャパシティ
- Shardの数を調整することによりどんな流量のStreamも処理をすることが可能です。
- ProducerはShardの値がいくつになっていても一意のエンドポイントにデータを流し込んでやればOKです。（Shardの数でキャパシティが決まりますので、適切な値に調整されている必要はありますが）
- 一方、 ConsumerはShardごとにホストやノード、またはプロセスを追加してやればよいでしょう。
これらの特性を活かすことにより、流量の遅いStreamから、非常に多数のProducer（例えば数十万もモバイルアプリケーション）から大量のデータがながれてくるようなStreamでも、アプリケーションを複雑化させることなく処理をすることができるでしょう。ここで紹介した概念を絵にすると以下のようになります。
Kinesisアプリケーションの設計と配置
また、ストリームプロセッシングを設計する上で非常に重要な概念となるのがConsumerアプリケーションの配置方法です。Amazon KinesisはストリームのMediatorのようなイメージで動作してくれます。データをストリームに流し込んでやれば、あとはMediatorであるAmazon Kinesisにコンセントを差し込むようなイメージで好きなConsumerアプリケーションを追加していくことができるのです。それぞれのアプリケーションは互いに影響をあたえることなく、同じデータを同じ順番（前述のSequence numberで実現されます）で受け取ることができます。
Webアプリケーションのログストリームを例にとると、以下の様な使い方が考えられます。
- Amazon Redshiftに逐次的にデータを流し込む
- リアルタイムダッシュボードで可視化する
- ユーザーごとのデータ解析結果をWebアプリケーションで利用しているDBに逐次フィードバックする
- 更に後日、別の解析処理を、上記の既存の処理に影響をあたえることなく追加する
データ入出力のエコシステム
これまで述べてきたような柔軟な構成をとるうえで非常に重要なのがAmazon Kinesisの入出力のためのライブラリです。ここから先のパートは、既に公開されているライブラリやツール群を(一部のツールはセットアップ方法等も交えながら)紹介していきます。下記の図はあくまでイメージですが、既に数多くのツール群が公開されています。
多くのものは、例えばFluentdのKinesisプラグインのように、既存のソフトウェアのプラグインとして実装されているので、既に利用中のソフトウェアに新たにライブラリやプラグイン、設定を追加するだけでAmazon Kinesisが利用できるようになります。
また、前述のFluentdのようにサーバー上で動作するものから、Amazon CognitoとAWS Mobile SDKのようにモバイルデバイス/モバイルデバイスアプリケーションからのデータ投入を実現してくれるものなどがありますので、様々な形のアプリケーションやシーンでAmazon Kinesisが利用可能です。
入力側ツール
Fluent Plugin for Amazon Kinesis
Fluentdはご存知の方も非常に多いのではないでしょうか。Treasure Dataによってスポンサーされているプロジェクトで、ログコレクタ/イベントコレクタのデファクトスタンダードと言ってもいいくらい普及しているツールです。インストールや設定方法についてはリンクをたどって頂ければと思いますが、アプリケーションサーバーなどのログ収集にFluentdを利用しているのであれば、このプラグインを追加するだけで簡単にAmazon Kinesisへもログを取り込むことが可能です。
セットップから利用までのイメージ
rubygems.orgで配布されていないので、githubから手動でダウンロードしてビルドする必要があります。
git clone https://github.com/awslabs/aws-fluent-plugin-kinesis
cd aws-fluent-plugin-kinesis
rake build
rake installapacheのアクセスログをtest-streamというストリームにデータを流し込む場合は以下のような設定ファイルを書くとよいでしょう。
<source>
type tail
path /var/log/httpd/access_log
tag apache
</source>
<match apache>
type kinesis
region ap-northeast-1
stream_name your-stream
partition_key your-key
flush_interval 1s
</match>
設定ファイルの詳細なオプションなどは後日別途取り上げたいと思います。
mqtt-kinesis-bridge
MQTTで受け取ったデータをAmazon Kinesisにフォワードしてくれるサーバーアプリケーションです。MQTTとは非常に軽量なメッセージングプロトコルで、IOTやM2Mの分野で、センサーデバイスなどからのデータ収集のための利用されることが期待されています。mqtt-kinesis-bridgeは、BrokerアプリケーションでであるMosquittoのバックエンドで動作し、MQTTで受け取ったデータをAmazon Kiesisに入力します。利用方法についてはGithub上のREADMEを参照してください。
Log4J Appender for Amazon Kinesis
Java開発者の方であればLog4Jを知らない方はいないのではないでしょうか。本ソフトウェアはLog4JのAppenderとして動作し、Amazon Kinesisへデータを入力してくれます。
上記のGithubプロジェクトからソースをダウンロードしてMavenを使ってビルドすることもできますが、利用したいだけならこちらのページからJarファイルをダウンロードできるようになっています。
出力側ツール
Amazon Kinesis Client Library for Java
Kinesis Client Library、もしくはKCLとして知られるこのソフトウェアはAmazonがファーストパーティーとして提供する、Consumerアプリケーションを開発するためのクライアントライブラリです。大規模なストリームを処理するためには複数のホストに分散したConsumerアプリケーションを協調して動作させる必要が出てきます。KCLはこのための処理のロードバランスや、ホストに問題が発生したときのフェイルオーバー、そのために必要となる処理のチェックポインティングなどをバックエンドで行ってくれます。これにより開発者は、Consumerアプリケーションのロジックの開発に多くのリソースを割くことができるようになります。
セットアップから利用までのイメージ
Maven Repositoryで配布されているので、Mavenプロジェクトから利用したいだけであれば以下のようにpom.xmlに追記してあげればOKです。
実際に利用するにあたってコアとなるのは、Consumerアプリケーションのメインロジックを実装することになる、IRecordProcessorを継承したクラスです。ここでinitialize, processRecords, shutdownという3つのインターフェイスを実装します。メソッド名からなんとなくイメージが湧くかと思いますが、processRecordsが実際にレコードを処理するためのメソッドで、ここに集計や分析などのロジックを書くことになります。サンプルコードをここに書くと長くなってしまうので、詳細を知りたい方はAWS Black Belt Techシリーズ Amazon Kinesisや、Developing Record Consumer Applications with the Amazon Kinesis Client Libraryなどを参照してください。実際に動くサンプルコードをみたほうが早いという方はAWS SDK for Javaに同梱されているサンプルコードや、Amazon Kinesis Data Visualization Sample Applicationを参照してください。
KCLについてはまた後日、このブログで個別に取り上げたいと思います。
Amazon Kinesis Connector Library
KCLと名前が非常に似ていますが別物です。こちらはAmazon DyanmoDB、Amazon Redshift、Amazon S3へのコネクタライブラリです。これを利用するとストリームのデータをリアルタイムに整形、フィルタしてRedshiftに格納する、などが実現できます。KCLと同様に、ConsumerアプリケーションとしてEC2などで動作させるタイプのライブラリです。詳細については後日、個別に取り上げたいと思います。
Amazon Kinesis Storm Spout
StormはTwitterが開発したストリームプロセッシングのためのアプリケーションフレームワークです。本ライブラリを利用することにより、Amazon KinesisをSpout（データの取り込み口）として使用できるので、Stormの経験や既存アプリケーション、トポロジなどの資産を有効に活用すうることができるでしょう。こちらについても別途個別に取り上げたいと思っています。
Amazon Elastic MapReduce
HadoopからもAmazon Kinesisのデータは利用可能です。EMRを最新のバージョンで起動すれば、すでにコネクタライブラリがインストールされた状態でクラスタが起動してきます。下記にHiveでストリームを取り扱うチュートリアルがありますので興味ある方はぜひ触ってみてください。
Tutorial: Analyzing Amazon Kinesis Streams with Amazon EMR and Hive
Apache Spark
Spark StreamingのデータソースとしてKinesisを利用することもできます。実際の設定方法などについては下記を参照してください。
Run Spark and Shark on Amazon Elastic MapReduce
Sparkとの連携についても、後日個別に取り上げたいと思っています。
まとめ
今回はAmazon Kinesis自体のおさらいと、ツール群のさわりを紹介させて頂きました。既に多数のツールが公開されていることが感じて頂けたのではないかと思います。次回以降は個別にツールを取り上げてセットアップ方法や実装のサンプルなどを紹介していきたいと思います。今のところ予定（あくまで予定です笑）として考えているのは以下のツールです。
- Amazon Kinesis Client Library
- Amazon Kinesis Connector Library
- Amazon Elastic MapReduce
- Apache Spark
- Fluentd Plugin for Amazon Kinesis
それでは次回以降をお楽しみに！