Amazon Kinesis Streams初期構築設定 2016年5月版

#AWS

quiver

2016.05.28

この記事は公開されてから1年以上経過しています。情報が古い可能性がありますので、ご注意ください。

AWSにはKinesis Streamsというフルマネージドなリアルタイムデータ処理サービスがあります。 Kinesis が必要とされるプロジェクトはなかなか触れる機会がなく、EC2やRDSのように一般的ではないため、運用の取っ掛かりのような情報がまとまっていません。

そこでAmazon Kinesis Streamsを利用するときに、最初にやっておくべき設定や重要な監視項目をまとめてみました。

なお Amazon Kinesis は以下の3サービスで構成されます。

Streams - ストリーミングデータを処理、分析する独自のカスタムアプリケーションを構築する
Firehose - 大量のストリーミングデータを AWS に簡単にロードできる(一部リージョンでのみ利用可能。東京はまだ利用できない)
Analytics - 標準 SQL でストリーミングデータを簡単に分析できる(Coming Soon)

今回の対象は Streams です。

Kinesis Streams の設定

データ保持期間の変更

デフォルトでは、ストリームに投入されたデータレコードは初期状態では24時間だけアクセス可能です。コンシューマーで障害が発生して処理が止まっていたり、コンシューマーの処理が遅延が積み重なると、あっという間にデータロストしてしまいます。

データロストしないように、データ保持期間を上限の7日まで延長しましょう。以下のコマンドを実行します。

$ aws kinesis increase-stream-retention-period \
  --stream-name foo \
  --retention-period-hours 168 # 168 = 24 hours/day x 7 days

保持期間の延長に伴い、Kinesis Streams の利用料は増えます。 Kinesis Streams は非常に安価に利用可能なため、Kinesis Streams を必要とするようなサービス全体の運用費を考えると、無視できる範囲かとおもいます。

参考リンク

https://dev.classmethod.jp/cloud/kinesis-increase-stream-retention/

シャード数の上限緩和

デフォルトではリージョンごとのシャード数は以下の通りです。

US East (N. Virginia)/US West (Oregon)/EU (Ireland) : 50
東京などその他のリージョン : 25

シャード数の制限はストリーム単位ではなくリージョン単位です。

東京リージョンで合計10シャードのストリームを運用していると、新規ストリームのシャード数は最大でも25-10=15です。

「レコードの処理が追いつかず、慌ててシャード分割しようとしたらシャード数上限に引っかかった！」なんてことが発生しないように、余裕を持った上限緩和申請を行い、十分なシャード数を確保しておきましょう。

拡張モニタリングの有効化

デフォルトではCloudWatchからはストリーム単位のメトリクスしか取得できません。拡張モニタリング機能を有効にすることで、シャード単位のメトリクスを取得できるようになります。

以下のコマンドを実行すると、すべてのシャードメトリクスが有効になります。

$ aws kinesis enable-enhanced-monitoring \
  --stream-name foo \
  --shard-level-metrics ALL

参考リンク

https://dev.classmethod.jp/cloud/kinesis-streams-supports-shard-level-metrics/

CloudWatch の重要なメトリクス

大前提として、シャード単位のメトリクスを有効にしてください。

Kinesis Streamsは機能(API)がシンプルなため、メトリクスの数は少なく、どのメトリクスも重要です。

しばらく運用したあとで有り難みを感じるメトリクスとして以下の3つを上げます。

(ストリーム・シャード)IteratorAgeMillisecondsのAVERAGE

この値は0に近いほど、リアルタイムにデータを処理出来ていることを意味します。

データ処理の止まってたり、データ処理遅延の発生しているシャード/ストリームをこのメトリクスのAVERAGEが増えます。

(シャードのみ)IncomingRecordsのSUM

ストリームを構成する各シャードでIncomingRecordsのSUMを取り、Producerが投入したレコードが各シャードに一様に分散されていることを確認します。

Producerが偏ったハッシュキーを設定してレコード投入していると、レコードは特定のシャードに偏より、シャード数を増やしてもうまくスケールアウトしません。

(ストリーム・シャード)ReadProvisionedThroughputExceeded/WriteProvisionedThroughputExceeded の SUM

処理されるレコードが少ないなぁと調べていたら、スロットルされていたなんてことが稀にあります。

このメトリクスは0が正常で、0 より大きい値が検出されたら、何か異常がおきています。

参考リンク

http://docs.aws.amazon.com/streams/latest/dev/monitoring-with-cloudwatch.html

Kinesisアプリケーションの対応

最後に Kinesis Streams そのものではありませんが、Kinesis アプリケーションについても触れます。

プロデューサー : PutRecordsでまとめてレコード投入

Kinesis Streamsにデータ投入するには、以下の2通りがあります。

PutRecord : 1 API呼び出しで1レコード追加
PutRecords : 1 API呼び出しで500レコードまで追加

最初から PutRecords API を使ってアプリケーションを構築し、スループットを上げましょう。

注意点としては、PutRecords API のリクエストが成功することと、すべてのレコードがストリームに追加されたことは別だということです。 APIのレスポンスには FailedRecordCount という属性が存在するため、一部のレコード処理で失敗した時は、そのリカバリー対応も必要です。

参考リンク

コンシューマー : LambdaのBatchSizeを大きくする

Kinesis Streams と Lambda を連携するケースは非常に多いかと思います。この2つを連携する場合、Lambda が呼び出されるたびに処理するレコード数の上限は Lambda 関数のイベントソースの BatchSize で制御しています。

デフォルトの BatchSize は 100 でかなり控えめに設定されています。 Lambdaを慣らし運転しながら、Lambdaの処理時間やメモリ使用量を眺めつつ、BatchSize を増やしてスループットを上げましょう。

BatchSize を変更するには以下のコマンドを実行します。 Lambda 関数の UUID を調べ、その UUID めがけて BatchSize を変更します。

$ aws lambda list-event-source-mappings --function-name foo
{
    "EventSourceMappings": [
        {
            "UUID": "99999999-dddd-4444-8888-222222222222",
            "StateTransitionReason": "User action",
            "LastModified": 1441078911.0,
            "BatchSize": 100,
            "State": "Enabled",
            "FunctionArn": "arn:aws:lambda:ap-northeast-1:111111111111:functionfooFood",
            "EventSourceArn": "arn:aws:kinesis:ap-northeast-1:111111111111:stream/Demo",
            "LastProcessingResult": "OK"
        }
    ]
}
$ aws lambda update-event-source-mapping \
  --uuid "99999999-dddd-4444-8888-222222222222" \
  --batch-size 500
{
    "UUID": "99999999-dddd-4444-8888-222222222222",
    "StateTransitionReason": "User action",
    "LastModified": 1441078933.47,
    "BatchSize": 500,
    "EventSourceArn": "arn:aws:kinesis:ap-northeast-1:111111111111:stream/Demo",
    "FunctionArn": "arn:aws:lambda:ap-northeast-1:111111111111:function:Food",
    "State": "Updating",
    "LastProcessingResult": "OK"
}