はじめに

リアルタイムストリーミングデータをS3やRedshiftに直接保存できるサービス、Amazon Kinesis Firehoseが発表されました。 【速報】データインジェスションサービス「Amazon Kinesis Firehose」がリリースされました！ #reinvent

Kinesisは大量のストリーミングデータを一時保存できるDBの一種ですが、一時保存されたデータを別の場所に写すにはユーザが自分でアプリケーションを作る必要がありました。 主なデータの移動先であるS3(永続化用)とRedshift(分析用)へのデータ転送機能をKinesisに追加したのがFirehoseです。

Kinesisの機能拡張であるFirehoseにはRedshiftの既存の課題を解決する側面があります。 Redshiftは大規模データ分析には欠かせないサービスです。しかし、特性上、個々のデータをINSERTにより登録しようとすると非常に時間がかかります。 そのため、ストリーミングデータの分析を行うにはデータを一時的にどこかに保存し、ある程度の量をまとめてロードするといった工夫をする必要がありました。 しかし、これからはFirehoseのDelivery Streamにデータを送れば、あとは自動でRedshiftにデータが登録されるようになります。

今回は、FirehoseでS3にデータを保存するところまでをやってみました。 保存先をRedshiftとした場合は、S3のデータをRedshiftにロードするところまでをFirehoseがやってくれます。

手順

FirehoseのDelivery StreamはKinesisの管理画面から作成します。 現在のところ、バージニア、オレゴン、アイルランドの3リージョンで利用可能です。 今回はバージニアに作りました。

保存先として、S3とRedshiftを選べます。今回はS3を選びます。

適切なIAM Roleを選択します。

Firehose用のRoleがない場合はその場で作成できます。 S3操作用の権限の他に暗号化のためKMSの権限も付与していることがわかります。

バッファーのルールや、圧縮暗号化の設定を行います。 バッファーはデータサイズと時間の両方で指定でき、どちらかの値に達した時点でKinesisからS3にデータが保存されます。 今回はどちらも小さめにしました。

確認画面です。

動作確認

それでは、実際にDelivery Streamにデータを送ってみます。 データの送信ツールはAmazon Kinesis AgentsとAWS SDKがありますが、今回はAmazon Kinesis Agentsを使います。 Amazon Kinesis Agentsの導入方法はこちらをご覧ください。 【速報】データインジェスションサービス「Amazon Kinesis Firehose」がリリースされました！ #reinvent

Agentの設定は次のようにしました。

[ec2-user@demo ~]$ cat /etc/aws-kinesis/agent.json
{
  "cloudwatch.emitMetrics": true,
  "cloudwatch.endpoint": "https://monitoring.us-east-1.amazonaws.com",
  "firehose.endpoint": "https://firehose.us-east-1.amazonaws.com",
  "flows": [
    {
      "filePattern": "/tmp/app.log*",
      "deliveryStream": "demo",
      "initialPosition": "START_OF_FILE"
    }
  ]
}

今回は省きましたがバッファーや複数行のログの扱いについても設定もできます。 詳細な設定項目はこちらを参照ください。 Writing to a Delivery Stream with Amazon Kinesis Agents

/tmp/app.logに文字を書き込みました。

[ec2-user@demo ~]$ cat /tmp/app.log
foo
bar
buz

しばらくしてから内容を追加しました。

[ec2-user@demo ~]$ echo add >> /tmp/app.log

Agentのログを確認してみます。

[ec2-user@demo ~]$ less /var/log/aws-kinesis-agent/aws-kinesis-agent.log
2015-10-08 01:29:52.287+0000 localhost (FileTailer[fh:demo:/tmp/app.log*].MetricsEmitter RUNNING) com.amazon.kinesis.str
eaming.agent.tailing.FileTailer [INFO] FileTailer[fh:demo:/tmp/app.log*]: Tailer Progress: Tailer has parsed 4 records (
16 bytes), and has successfully sent 3 records to destination.
2015-10-08 01:29:52.292+0000 localhost (Agent.MetricsEmitter RUNNING) com.amazon.kinesis.streaming.agent.Agent [INFO] Ag
ent: Progress: 4 records parsed (16 bytes), and 3 records sent successfully to destinations. Uptime: 90037ms
2015-10-08 01:30:22.286+0000 localhost (FileTailer[fh:demo:/tmp/app.log*].MetricsEmitter RUNNING) com.amazon.kinesis.str
eaming.agent.tailing.FileTailer [INFO] FileTailer[fh:demo:/tmp/app.log*]: Tailer Progress: Tailer has parsed 4 records (
16 bytes), and has successfully sent 3 records to destination.
2015-10-08 01:30:22.291+0000 localhost (Agent.MetricsEmitter RUNNING) com.amazon.kinesis.streaming.agent.Agent [INFO] Ag
ent: Progress: 4 records parsed (16 bytes), and 3 records sent successfully to destinations. Uptime: 120037ms
2015-10-08 01:30:52.286+0000 localhost (FileTailer[fh:demo:/tmp/app.log*].MetricsEmitter RUNNING) com.amazon.kinesis.str
eaming.agent.tailing.FileTailer [INFO] FileTailer[fh:demo:/tmp/app.log*]: Tailer Progress: Tailer has parsed 4 records (
16 bytes), and has successfully sent 4 records to destination.

なんとなく送られていそうです。

S3バケットを確認します。 オブジェクトが2つ作られています。 中身を確認したところ、１つ目は

foo
bar
buz

2つ目は

add

と問題なくデータが送られていました。

おわりに

今回は少しのデータしか送りませんでしたが、裏はKinesisなのでかなりの流量のデータを送ることができます。 とりあえず、Firehoseにデータを流しておけば、用途によってS3に貯めたり、Redshiftで分析したりと便利に使えそうです。