S3のイベントトリガーでのLambda発火についての実験

#S3のイベントトリガーでのLambda発火についての実験

内容

s3へファイル(オブジェクト)を保存した時に、Lambdaを発火させて、ゴニョゴニョする。というのは、よくあるアーキテクチャの1つです。
この場合は、Lambdaは「At Least Once」(最低1回)で起動するというのは周知のことですが、
「(LambdaはAt Lease Onceだけど)そもそもLambdaを発火させるため、S3のイベント自体が発行されないことがある」と聞いたので、それを実験してみました。

事前情報

こちらにも「イベント抜け」と記載してあります
https://qiita.com/kkimura/items/92833436349d8fc5ad29

やってみる

1. EC2上に空の100万ファイル作成
2. s3バケット作成
3. DynamoDBのテーブル作成
4. Lambda作成とトリガー設定
5. aws s3 syncで100万ファイルをs3へコピー
6. 結果確認

EC2上に空の100万ファイル作成

20秒位で100万ファイル作成できました

$ mkdir s3-test
$ cd s3-test
$ time seq 1 1000000 | xargs touch 

s3バケット作成

別になんでもいいですが、「s3-test-qiita」というバケットを作成し、
「files」というフォルダ内に作成したファイルをコピーするようにします。

DynamoDBのテーブル作成

テーブル名もなんでもいいですが、「s3-test-qiita」でパーティションキーを「key(string)」としています。
s3 sync実行直前にRCU/WCUを調整したいため、どちらもautoscalingなしで「1」としています。

Lambda作成とトリガー設定

python3.6で以下の通り書きました。メモリ割り当ては最小ですが、
DymamoDBのスロットリングでリトライが発生する場合があるため、Timeoutは1分にしています。
s3のイベントタイプは「Object Created(All)」としています。

# -*- coding: utf-8 -*-

import json
import boto3

dynamodb = boto3.resource('dynamodb')

DDB_TABLE = "s3-test-qiita"
table = dynamodb.Table(DDB_TABLE)

def lambda_handler(event, context):
    bucket = event['Records'][0]['s3']['bucket']['name']
    key = event['Records'][0]['s3']['object']['key']

    table.update_item(
        Key= {
            'key': key
            },
        UpdateExpression="ADD #name :increment",
        ExpressionAttributeNames={
            '#name':'count'
            },
        ExpressionAttributeValues={
            ":increment": 1
            },
        ReturnValues="UPDATED_NEW"
    )

動作確認テスト(必要であれば)

作成したs3バケットにファイルを保存して、DynamoDBにレコードが作成されていることを確認します。
さらに、同じファイルを再度保存(上書き)して、DynamoDBのレコードの「count]がインクリメントされていることを確認します。

EC2からs3 syncの実行

その前にDynamoDBのWCU/RCUを上げます。検証終わり次第すぐに消すので、一時的にそれぞれ500に変更しました。

$ cd ../
$ aws s3 sync s3-test/ s3://s3-test-qiita/files/
upload: s3-test/10279 to s3://s3-test-qiita/files/10279          
upload: s3-test/102792 to s3://s3-test-qiita/files/102792        
upload: s3-test/102793 to s3://s3-test-qiita/files/102793        
upload: s3-test/102791 to s3://s3-test-qiita/files/102791        
upload: s3-test/102794 to s3://s3-test-qiita/files/102794        

大体1時間かかりました。DynamoDBのWrite capacityを見ると、250を常に消費しているようでした。
つまり、1秒あたり250ファイルをputできたようです。ただし、s3へのPUTは小さいファイルがあるなら、圧縮する。並列で実行するなどベストプラクティスがあるので、これでS3が早いとか遅いとかの話をすべきじゃないですね。
あと、Lambdaはスロットリングもエラーも発生してないことはcloudwatchから確認済みです。

結果確認

さて、結果の確認方法ですが、DynamoDBには集計機能がないので、個人的な勉強も含めてEMRでやってみたいと思います。
他には、Scanで全レコード取り出して集計(1MB制限に注意)やdatapipelineでs3に吐いて集計(フォーマット注意)などありますね。

EMRを起動後、SSH接続してhiveを実行

外部テーブルの作成

[hadoop@ip-172-31-43-65 ~]$ hive
hive> 
CREATE EXTERNAL TABLE hivetable (col1 string, col2 bigint)
STORED BY 'org.apache.hadoop.hive.dynamodb.DynamoDBStorageHandler' 
TBLPROPERTIES ("dynamodb.table.name" = "s3-test-qiita", 
"dynamodb.column.mapping" = "col1:key,col2:count"); 

総レコード数の算出

hive> select count(col1) from hivetable;
Query ID = hadoop_20180502104919_f1b374a8-cdd5-4668-a110-2851e0e9fbae
Total jobs = 1
Launching Job 1 out of 1
Tez session was closed. Reopening...
Session re-established.
Status: Running (Executing on YARN cluster with App id application_1525254801764_0002)

----------------------------------------------------------------------------------------------
        VERTICES      MODE        STATUS  TOTAL  COMPLETED  RUNNING  PENDING  FAILED  KILLED  
----------------------------------------------------------------------------------------------
Map 1 .......... container     SUCCEEDED      2          2        0        0       0       0  
Reducer 2 ...... container     SUCCEEDED      1          1        0        0       0       0  
----------------------------------------------------------------------------------------------
VERTICES: 02/02  [==========================>>] 100%  ELAPSED TIME: 17.15 s    
----------------------------------------------------------------------------------------------
OK
1000000
Time taken: 26.7 seconds, Fetched: 1 row(s)

お、1,000,000レコードとなりました！！！少なくとも今回の検証では、イベント通知抜けというのはなかったようです。

Lambda複数回起動の特定

hive> select col2, count(col2) from hivetable group by col2;
Query ID = hadoop_20180502105243_ab71ff5d-2493-4c58-8e5b-2bea7d657c0d
Total jobs = 1
Launching Job 1 out of 1
Status: Running (Executing on YARN cluster with App id application_1525254801764_0002)

----------------------------------------------------------------------------------------------
        VERTICES      MODE        STATUS  TOTAL  COMPLETED  RUNNING  PENDING  FAILED  KILLED  
----------------------------------------------------------------------------------------------
Map 1 .......... container     SUCCEEDED      2          2        0        0       0       0  
Reducer 2 ...... container     SUCCEEDED      1          1        0        0       0       0  
----------------------------------------------------------------------------------------------
VERTICES: 02/02  [==========================>>] 100%  ELAPSED TIME: 15.10 s    
----------------------------------------------------------------------------------------------
OK
1       999988
2       12
Time taken: 16.221 seconds, Fetched: 2 row(s)
```
お、100万回のうち、999988はLambdaが1回だけ起動し、12回はLambdaが2回起動してしまったようです。
1回のPUTで、s3のイベントが複数回通知されたのか、イベント通知は1回だけどもLambdaが2回起動したのかは不明です


## 結果
少なくとも今回の検証では、100万オブジェクトをPUTしても、Lambdaは抜けなく起動した。
ただし、少なくとも抜けが起こることは確認されているので、絶対に抜けが許容できないシステムなら別途対策が必要です。(その話はまた後日) その対策がめんどくさいなら、「抜けが報告された時点で手動で対応する」と決めるのもありです。