Node.jsのStreamについて知らなかったのでAWS SDKのS3のファイル読み出しをしながら使ってみた時のメモ

参考

以下を読ませていただきました。

また、以下の本のStreamの欄を見てみました。

サーバサイドJavaScript Node.js入門

Streamとは

データの流れを扱う際に利用する抽象的なインターフェース
例えばファイルを読み込む場合、すべてのファイルの内容をメモリに持ってから処理するのではなく、少しずつ読み込みながら逐次処理を行う事で効率化できる。Node.jsっぽい
Readable StreamでsetEncodingしないとBufferオブジェクトとして読み込む。文字列として読み込みたい場合にはreadStream.setEncoding('utf-8')などする。(Bufferはバイナリを扱うためのクラス)

やってみる

まず、普通にS3からELBのアクセスログ情報を取得してみます。
以下のようなコードを書きます

s3.js

var AWS = require('aws-sdk');

var params = {
  Bucket: 'toshihiro-elb-test',
  Key: 'AWSLogs/xxxxx/elasticloadbalancing/ap-northeast-1/2015/05/29/694273932022_elasticloadbalancing_ap-northeast-1_LoadBlancer_20150529T1315Z_52.68.91.81_3jjesltj.log'
}

var s3 = new AWS.S3();
s3.getObject(params, function(err, data) {
  if (err) console.log(err, err.stack);
  else console.log(data);
});

実行してみます。

$node s3.js
{ AcceptRanges: 'bytes',
  LastModified: 'Fri, 29 May 2015 13:19:28 GMT',
  ContentLength: '3138',
  ETag: '"c6c596d459088a4a6017f9db3570a216"',
  ContentType: '',
  Metadata: {},
  Body: <Buffer 32 30 31 35 2d 30 35 2d 32 39 54 31 33 3a 31 33 3a 35 37 2e 34 35 30 38 36 37 5a 20 4c 6f 61 64 42 6c 61 6e 63 65 72 20 31 30 31 2e 31 31 31 2e 32 30 ... > }

上記コードの場合、一度ファイルのすべての内容をメモリに読み込んで処理をする形になります。
また、Body部分はBufferとして取れているようです。

上記Body部分（ファイルの中身）をReadableStreamを使って非同期で読み込む事をやってみます。

s3.js

var AWS = require('aws-sdk');

var params = {
  Bucket: 'toshihirock-elb-test',
  Key: 'AWSLogs/xxxxxxxxxx/elasticloadbalancing/ap-northeast-1/2015/05/29/xxxxxxxx_elasticloadbalancing_ap-northeast-1_LoadBlancer_20150529T1315Z_52.68.91.81_3jjesltj.log'
}

var s3 = new AWS.S3();
var s3Stream = s3.getObject(params).createReadStream();

s3Stream.on('data', function(chunk) {
  console.log(chunk);
});

s3Stream.on('end', function() {
  console.log('end');
});

実行してみます

$node s3.js
<Buffer 32 30 31 35 2d 30 35 2d 32 39 54 31 33 3a 31 33 3a 35 37 2e 34 35 30 38 36 37 5a 20 4c 6f 61 64 42 6c 61 6e 63 65 72 20 31 30 31 2e 31 31 31 2e 32 30 ... >
<Buffer 61 6d 61 7a 6f 6e 61 77 73 2e 63 6f 6d 3a 38 30 2f 66 61 76 69 63 6f 6e 2e 69 63 6f 20 48 54 54 50 2f 31 2e 31 22 20 22 4d 6f 7a 69 6c 6c 61 2f 35 2e ... >
<Buffer 35 33 37 2e 33 36 20 28 4b 48 54 4d 4c 2c 20 6c 69 6b 65 20 47 65 63 6b 6f 29 20 43 68 72 6f 6d 65 2f 34 33 2e 30 2e 32 33 35 37 2e 38 31 20 53 61 66 ... >
<Buffer 35 33 37 2e 33 36 20 28 4b 48 54 4d 4c 2c 20 6c 69 6b 65 20 47 65 63 6b 6f 29 20 43 68 72 6f 6d 65 2f 34 33 2e 30 2e 32 33 35 37 2e 38 31 20 53 61 66 ... >
end

最初の時と同様、Bufferとして読み込んでいるようです。
読み込む際にUTF-8として読み込むように以下の記述を追記します。

+ s3Stream.setEncoding('utf-8');

実行してみます。

$node s3.js
2015-05-29T13:13:57.450867Z LoadBlancer 
...

文字列として取得し、表示できました。

ただし、上記dataイベント発生時にchunkで取得できる文字列は複数行の文字列となり、１行ごとに処理を行う場合には別途対応が必要です。(ログを1行ごとに読み込んでRDSに登録するなど）

試しにdataイベントが呼ばれる都度、aaaaaと出力するようにすると以下のように複数行の文字列がdataイベント時にきているのが分かります。

2015-05-29T13:13:57.450867Z LoadBlancer 
...
aaaaaa
amazonaws.com:80/favicon.ico HTTP/1.1" "Mozilla/5.0 - -
...

chunkで受け取った時に改行コードをみてごにょごにょやれば良いのですが、面倒なので byline というnpmから提供されるLineStreamクラスを利用して1行ごとに読み込めるようにします。まずは、npmコマンドでモジュールをインストールします。

jahewson/node-byline

$npm install byline

コードを変更します。
一度LineStreamをpipeしてからdataイベントを受け取る事で1行ごとの処理を行う事ができます。

s3.js

var AWS = require('aws-sdk');
var LineStream = require('byline').LineStream;

var lineStream = new LineStream();

var params = {
  Bucket: 'toshihirock-elb-test',
  Key: 'AWSLogs/xxxxxxx/elasticloadbalancing/ap-northeast-1/2015/05/29/xxxxxxxx_elasticloadbalancing_ap-northeast-1_LoadBlancer_20150529T1315Z_52.68.91.81_3jjesltj.log'
}

var s3 = new AWS.S3();
var s3Stream = s3.getObject(params).createReadStream();
s3Stream.setEncoding('utf-8');

s3Stream
  .pipe(lineStream)
  .on('data', function(line) {
    console.log(line);
    console.log('aaaaaa');
  });

s3Stream.on('end', function() {
  console.log('end');
});

上記を実行すると１行ごとに読み込めている事が分かります。
この方法を使えば一度にファイルの内容を読み込まず、非同期でファイル1行ごとの処理を簡単にかけて重宝しそうです。Lambdaでも使うとよさげです。

AWS SDKでNode.jsのReadableStreamを使ってみる

参考

Streamとは

やってみる