Serverless FrameworkでAWS Lambda関数作ってRDSのデータで色々レポーティングする

はじめに

• 少し使ってるバージョンが古いです(Node.jsのバージョン上げれば大丈夫なはずだけで環境用意してなくて未実施。)
• ソースコードは書き殴ったままなので見る人が見たらアレルギー反応出ちゃうかもしれません
• 年末大掃除の1つ。

動作環境、使ってるもの

• Serverless Framework
• AWS Lambda
• Node.js 8.10 (現時点でLambda関数が新規作成できないです)

なぜ作ったか

SQLをプログラム中で組むのが面倒だった。
集計用のSQLは重くなりがちなので、チューニングしやすいようにSQLそのもので管理したかった。
依頼されているレポートをまとめて作りたかった。
レポートを10前後(重いのは2つ程度)を毎日作成することから解放されたかった。

仕組み

• 指定のディレクトリに格納されているすべてのSQLを実行する
• 実行結果をCSVファイルに書き込む
• (記事中に載せてない処理)
  • すべてのCSVファイルをzipファイルにまとめる
  • zipファイルをSlackに投稿

ソースコード

GetRdsData.js

'use strict';

/**
 * 1回の取得件数、繰り返しの回数を指定。
 */
const ONETIME_ROWS = 200000; // 1回あたりの取得件数 大きすぎるとメモリが足りないのでほどほどに。
const GET_TIMES = 20; // 取得回数 ONETIME_ROWSの値と合わせて指定。
const START_OFFSET_CNT = 0; // 最初から取得する場合は0を指定。

/**
 * @namespace GetRdsData
 */

// ライブラリ読み込み
const AWS = require('aws-sdk');
const wrapper = require('co-mysql');
const mysql = require('mysql');
const co = require('co');
const fs = require('fs');
const del = require('del');
const dateformat = require('dateformat');
const Json2csvParser = require('json2csv').Parser;

const log = function() {
    console.log.apply(console, arguments);
};

/**
 * MySQL 接続設定
 * @memberof GetRdsData
 * @type {Object}
 */
const opts = {
    connectionLimit: 1,
    host: process.env.RdsEndpoint || '127.0.0.1',
    port: process.env.RdsPort || '3306',
    user: process.env.RdsUser || 'user',
    password: process.env.RdsPassword || 'password',
    database: process.env.RdsDbName || 'db',
    multipleStatements: true
};

const STAGE_LABEL = process.env.Environment;

/**
 * SQLファイル格納先
 *
 *
 * @memberof GetRdsData
 */
const SQL_FILE_DIR = './sql/';
const TMP_DIR = '/tmp/';
const RDS_FILE_PREFIX = 'rds-';

const outputZipFileName = RDS_FILE_PREFIX + STAGE_LABEL + '_' + dateformat(new Date(), 'yyyymmddHHMMss') + '.zip';
const outputZipFilePath = TMP_DIR + outputZipFileName;

/**
 *
 * データ取得、ファイル作成 <br />
 *
 *
 * @memberof GetRdsData
 * @name GetRdsData
 * @param  {event}   event    [description]
 * @param  {context}   context  [description]
 * @param  {Function} callback [description]
 * @return {}            [description]
 */
exports.handler = (event, context, callback) => {
    log(opts);

    const executeQuery = function*(file, start, limit) {
        log('executeQuery', file);

        // SQLファイル読み込み
        const sql = fs.readFileSync(SQL_FILE_DIR + file, 'utf-8');

        // 参照系(SELECT,SET)以外を除外
        const checkStr = sql.split(";");
        for (let i = 0; i < checkStr.length; i++) {
            if ('SELECT' !== checkStr[0].slice(0, 6).toUpperCase() && 'SET' !== checkStr[0].slice(0, 3).toUpperCase()) {
                log('sql check NG!!!');
                return null;
            }
        }

        // SQL実行
        const result = yield p.query(sql, [start, limit]);

        if (result == null || result.length == 0) {
            return null;
        }

        log('sql execute ok');

        // SQLの実行結果で分岐
        if (result[0].constructor.name == 'RowDataPacket') {
            // SQLが単独の場合はそのまま返却
            return result;
        } else {
            // SQLが複数の場合は末尾の結果のみを返却
            return result[result.length - 1];
        }
    };

    /**
     * CSVファイルを出力する
     * 出力先はtmpディレクトリ
     */
    const outputTmpCsvFile = function*(fileName, object) {
        // Excel文字化け対策のためにUTF-8 BOM付きで保存する
        fs.writeFileSync(TMP_DIR + fileName, '\ufeff' + object);
    };

    const pool = mysql.createPool(opts);
    const p = wrapper(pool);
    const ymdMap = new Map();

    co(function*() {

        // SQLファイル一覧を取得
        const fileList = fs.readdirSync(SQL_FILE_DIR);

        // SQLをすべて実行する
        for (let file of fileList) {
            // SQLファイル以外はスキップ
            if (!fs.statSync(SQL_FILE_DIR + file).isFile() || file.indexOf('.sql') < 0) {
                continue;
            }

            log(file);

            for (let i = START_OFFSET_CNT; i < START_OFFSET_CNT + GET_TIMES; i++) {
                let start = ONETIME_ROWS * (i);
                let end = ONETIME_ROWS * (i + 1); // SQL実行には使わない

                // SQL実行
                const result = yield executeQuery(file, start, ONETIME_ROWS);

                log('data convert csv')

                // データの取得結果で分岐
                if (result != null && result.length > 0) {
                    log(result.length);
                    // データが取得できた場合
                    // 個別集計のデータの場合
                    const json2csvParser = new Json2csvParser({});
                    const csv = json2csvParser.parse(result);

                    log('convert done. file output start.')

                    // データの出力
                    yield outputTmpCsvFile(RDS_FILE_PREFIX + file.replace('.sql', '') + '_' + start + 'to' + end + '.csv', csv);
                    log('file output done.')
                } else {
                    // データが取得できなかった場合
                    log('get data error', file);
                    log(result);
                    break;
                }
            }
        }

        // Connection Poolの解放
        pool.end();

    }).then(() => {
        // tmpディレクトリ下に作成したファイルを削除
        // csvファイル削除
        del.sync([TMP_DIR + RDS_FILE_PREFIX + '*.csv'], {
            force: true
        });
    }).catch((e) => {
        log(e);
        callback(e, null);
    });
}

package.json

{
    "name": "get-rds-data",
    "version": "0.0.1",
    "private": true,
    "engines": {
        "node": "8.10.0"
    },
    "main": "GetRdsData.js",
    "dependencies": {
        "co": "*",
        "co-mysql": "*",
        "mysql": "*",
        "json2csv": "^4.5.1",
        "del": "*",
        "request": "*",
        "archiver": "*",
        "dateformat": "*",
        "nodejslack": "*"
    },
    "devDependencies": {
        "aws-sdk": "*"
    },
    "scripts": {}
}

serverless.yml

service: getRdsData

provider:
  name: aws
  runtime: nodejs8.10
  memorySize: 2048
  timeout: 120
  region: ap-northeast-1
  stage: ${opt:stage, 'develop'}
  environment:
    TZ: Asia/Tokyo
    Environment: ${self:provider.stage}

resources:
  Description: Get data.

# 環境依存の設定値

custom:
  stage:
    # ローカル実行
    local:
      RdsEndpoint: 127.0.0.1
      RdsPort: 3306
      RdsDbName: db
      RdsUser: user
      RdsPassword: password

# 機能固有の設定値

functions:
  GetRdsData:
    handler: GetRdsData.handler
    description: Get data.
    role:
    environment:
      RdsEndpoint: ${self:custom.stage.${self:provider.stage}.RdsEndpoint}
      RdsDbName: ${self:custom.stage.${self:provider.stage}.RdsDbName}
      RdsUser: ${self:custom.stage.${self:provider.stage}.RdsUser}
      RdsPassword: ${self:custom.stage.${self:provider.stage}.RdsPassword}
      RdsPort: ${self:custom.stage.${self:provider.stage}.RdsPort}

まとめ

だいたいどこ行ってもSQLを簡単に実行できる仕組みは必要だった。