Kinesis Firehoseで送るデータをLambdaで整形するNode.jsの実装方法

どうも高妻です。Twitterでプログラミングのことや暗号通貨のことをつぶやいているのでよかったらフォローしてください。Follow @Tomokazu106

Lambdaでログ整形

Kinesis FirehoseでログをS3に転送し、そのログをLambdaで受け取りログを整形する場合を想定しています。そのときのLambdaの処理をNode.jsでどう書くか説明します。

ソースコード

exports.handler = (event, context, callback) => {
    const output = event.records.map(record => {
        const data = Buffer.from(record.data, 'base64').toString('utf8');
        const parsedData = JSON.parse(data);

        parsedData.text = "textを追加"

        return {
            recordId: record.recordId,
            result: 'Ok',
            data: Buffer.from(parsedData, 'utf8').toString('base64')
        };
    });
    callback(null, {records: output});
};

exports.handler = (event, context, callback) => {

const output = event.records.map(record => {

const data = Buffer.from(record.data, 'base64').toString('utf8');

const parsedData = JSON.parse(data);

parsedData.text = "textを追加"

return {

recordId: record.recordId,

result: 'Ok',

data: Buffer.from(parsedData, 'utf8').toString('base64')

};

});

callback(null, {records: output});

};

上記がデータを受け取ってから返すまでの処理です。ES6で書いているのでBabelでトランスパイルしてからソースコードをアップしてください。関数名をhandlerとしていますがどんな名称を使用しても大丈夫です。

トランスパイルは下記記事を参考にしてください。

ES6をトランスパイルしてAWS Lambdaで使用する方法

データの受け取り方

第一引数のevent.recordsから取得できます。Kinesis Firehoseから受け取るデータは複数になります。

データはbase64エンコードされたものが渡されるのでこれをutf8に変換してあげないといけません。またJSONで送っている場合はJSON.parseもしないといけません。

まとめると今回のデータ変換の流れは下記です。

Binary(base64)-> String(UTF8) -> JSON

1	Binary(base64)-> String(UTF8) -> JSON

データの返し方

データを返すときは変更したdataを上書きし、resultキーに’Ok’という文字列を設定します。最後に受け取ったときと同じ状態に戻すためにbase64エンコードでバイナリデータに変換し、callbackの第二引数に渡してあげれば完了です。

おすすめ本

初めてAWSでサービス構築する方はこの本がおすすめです。IAMを使ったユーザ管理からVPCやセキュリティグループの設定、EC2・RDSなどを使った基本的なインフラ構築が丁寧に解説されています。初心者の入門本としておすすめの一冊になります。

Amazon Web Services実践入門

posted with ヨメレバ

舘岡守/今井智明技術評論社 2015年11月10日

Amazon

Kindle

楽天ブックス

この記事を書いた人
最新の記事

高妻智一

2013年CyberAgent新卒入社スマホゲームを作る子会社に所属し、サーバーサイドのエンジニアを担当。２年目の終わりから新規子会社の立ち上げに参加し、サーバーサイドのエンジニアリーダーとしてサービースのリリースから運用までを担当。 2018年仮想通貨のスマホウォレットを提供するGinco Incにブロックチェーンエンジニアとして入社。