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を簡単に実行できる仕組みは必要だった。