S3向けLambda関数の既存ファイルに対する繰り戻し実行を考える(レジューム付き)

S3にファイルが作成または更新されたときLambdaで何か処理(画像変換とか)をするというのはLambdaのユースケースのひとつです。

問題になるのは、新しく作られたファイルには自動でLambda関数が実行されますが、すでに存在していたオブジェクト(以降ファイルと呼びます)にどう実行をするかという点です。これを繰り戻し実行と表現しました。

できるだけ既存のコマンドで汎用的に使える定石を考えてみます。処理の中断と再開も考慮します。

# 前提

次のリソースに対する処理を前提にします。お使いのS3バケットや関数で読み替えてください。

* S3バケット `BUCKET`
* プレフィックス(起点ディレクトリ) `PREFIX`
* サフィックス(対象の拡張子) `.jpg`または`.jpeg` ※ 他の形式の場合は読み替えてください。
* Lambda関数名 `FUNCION`

# まずは対象ファイルのリストアップ作業

まずはS3のファイルパスをリストアップする方法です。`todo.txt`ファイルにパス一覧を出力します。

`aws`コマンドの高レベルコマンドを使った方が安心なので、こんな風にしました。

```bash
aws s3 ls --recursive s3://BUCKET/path \
  | perl -n -e 'print [split(/\s+/, $_, 4)]->[3] if /\.jpe?g$/i' > todo.txt
```

lsは時刻やファイルサイズ、ディレクトリに相当する行も出力してしまい、それを省略する方法がないので、`awk`で`.jpe?g`のみフィルタリングしつつ、ファイルパス部分を切り出しています。

はじめは`awk`を使いましたが、スペース(複数かも知れない)を含むファイル名の抽出で躓いたのでインストール済みであることが多い`perl`にしました。

`ruby`でも簡単です。

```bash:rubyの場合
aws s3 ls --recursive s3://BUCKET/path \
  | ruby -n -e 'print $_.split(/\s+/, 4)[3] if /\.jpe?g/i' > todo.txt
```

# S3向けのLambdaイベントの人為的な発火について

Lambda関数はあくまで関数なので、同じ引数を与えて実行すれば、それが実際のイベントによるものか人為的に呼び出されたものか区別されません。

そのようにLambda関数を外部から起動するには、`aws`コマンドの`lambda invoke`サブコマンドを使います。

`--payload`に`event`引数で参照される部分をJSON形式で指定すると、人為的にS3のオブジェクトPOSTイベントを発火できます。

次は、あるひとつのオブジェクト`KEY`が作成または更新されたことにしてLambda関数を実行する例です。

```bash
aws lambda invoke \
      --invocation-type RequestResponse \
      --function-name FUNCTION \
      --payload '{"Records":[{"s3":{"bucket":{"name":"BUCKET"},"object":{"key":"KEY"}}}]}' \
      ./result.txt
```

# まとめて一気に実行

`todo.txt`に出力されたパスすべてにLambda関数を実行します。
`xargs`でひとつずつもありですが、せっかくのLambdaなので並列に実行します。

## parallelコマンドによる並列実行

コマンドラインで並列処理を実行できる、`parallel`コマンドを使う方法です。

### インストール

[GNU Parallel](http://bicycle1885.hatenablog.com/entry/2014/08/10/143612)をインストールしてください。

#### macOSの場合

```bash
brew install parallel
```

#### Ubuntuの場合

```bash
sudo apt install -y parallel
```

### 並列実行

上記で`todo.txt`に出力されたパスの一覧に対し、最大10プロセス同時にLambda関数を起動します。完了したパスを逐次`done.txt`に追記していきます。

```bash
parallel -j 10 -a todo.txt \
  'aws lambda invoke --invocation-type RequestResponse \
    --function-name FUNCTION --payload \
    '\''{"Records":[{"s3":{"bucket":{"name":"BUCKET"},"object":{"key":"{}"}}}]}'\'' \
    /dev/null; echo "{}" >> done.txt'
```

次のような出力がずらずらと出てきます。非表示にする方法が不明でした(標準エラーを`/dev/null`に捨てても表示されてしまう…)。

```
{
    "ExecutedVersion": "$LATEST", 
    "StatusCode": 200
}
{
    "ExecutedVersion": "$LATEST", 
    "StatusCode": 200
}
```

`-j 10`の数値が同時実行数です。やはりプロセス実行のオーバーヘッドが大きく、手元の環境では`50`にするとCPUが上限に張り付いてしまいました。

## Rubyによる並列実行

APIコールの度に毎回、`aws`コマンドのプロセスを起動するのは無駄なので簡単なRubyでスクリプトを組みます。

こちらだと手元のMacとWiFi接続で並列200で安定して処理できました。すごいレバレッジ。

ファイルを読みこんで並列にAWSのSDKを実行するだけなので、他の言語への移植も難しくないと思います。

### aws-sdkとparallelモジュールをインストール

はじめにGemモジュール`aws-sdk`と`parallel`をインストールします。

```bash
gem install aws-sdk parallel
# システムRubyの場合は sudo install aws-sdk parallel
```

ワンライナーはそろそろ厳しいので、`run.rb`ファイルを作成します。`BUCKET`と`FUNCTION`はお使いのもので差し替えてください。

```ruby:run.rb
require 'aws-sdk'
require 'parallel'

the_bucket='BUCKET'
the_function='FUNCTION'
parallel = 200

client = Aws::Lambda::Client.new
m = Mutex.new

File.write('done.txt', '')
Parallel.each(IO.readlines('todo.txt', chomp: true), in_threads: parallel) do |path|
  begin
    json = %Q|{"Records":[{"s3":{"bucket":{"name":"#{the_bucket}"},"object":{"key":"#{path}"}}}]}|
    res = client.invoke(function_name: the_function, payload: json)
    if res[:status_code] == 200
      begin
        m.lock
        File.write('done.txt', path + "\n", mode: 'a')
      ensure
        m.unlock
      end
    else
      puts "#{path} status:#{res[:status_code]}"
    end
  rescue => e
    puts path
    p e.message
  end
end
```

このスクリプトを実行すると、`parallel`コマンドと同様に実行されたパスが`done.txt`ファイルに逐次追記されていきます。

```bash
ruby run.rb
```

# レジューム: 処理の再開

途中で止まってしまったとき、またはじめから実行するのは無駄なので未処理のファイルのみを対象にします。

対象の`todo.txt`のうち、`done.txt`に含まれないパスを抽出して`todo.txt`に保存しなおします。これを`sort`と`comm`コマンドで行います。

```bash
sort todo.txt -o todo.txt; sort done.txt -o done.txt; \
  comm -2 -3 todo.txt done.txt > tmp.txt; mv tmp.txt todo.txt
```

`comm`コマンドを`> todo.txt`に直接リダイレクトしてしまうと出力が空になってしまうので`tmp.txt`を経由します。

これで`done.txt`に記録された処理済みのパスが、`todo.txt`から除外されます。また`todo.txt`に記録されたパスに対して処理を行えば、再開に当たるというわけです。