Firebase RealtimeDatabaseに大量データ投入 - その2 -

GYAOのtsです。

#経緯
前回の投稿の設計を実際に構築してみる。

#やりたいこと
前回の投稿参照

• ユーザーに対するお知らせをメッセージとして格納していく。
• 全ユーザーに対するお知らせ、個別ユーザーに対するお知らせがある。
• メッセージの変更、削除ができる
• 個別ユーザーに対するお知らせは1回で500万程度のユーザーに対して。

#Database Rule

.rules

{
  "rules": {
    "users": {
      "$uid": {
        ".write": "$uid === auth.uid",
        ".read": "$uid === auth.uid"
      }
    },
    "messages": {
      ".read": true,
      ".write": false,
      ".indexOn": ["expireDate"]
    }
  }
}

• users
  • Firebaseが発行したIDを使用（結局うら側でIDは紐付け）するので、自分のお知らせしか参照できないようにする。既読未読も管理するため、自分のお知らせをupdateできるように。
• messages
  • メッセージに関してはfunctionsから登録されるため、読み取り権限のみ。ユーザーは書き込めない。
• index
  • orderByChild("expireDate")で一気に期限切れメッセイジを検索してremoveするため、expireDateにindexを設定。

#関数upsertJsonの作成
CloudFunctionsのevent triggerでバケットを監視。uploadを検知し
CloudStorageのpathがmessages/*ならmessagesに、users/*ならusersにupsert（マージ）する。

##firebase init
下記を参考にfirebase init functionsでディレクトリ作成
https://firebase.google.com/docs/functions/get-started

###ディレクトリ構成

（cloudbuild.yamlは追加しました　。）

##coding

index.js

'use strict';

process.env.FIREBASE_CONFIG = JSON.stringify({
    databaseURL: 'https://xxxx.firebaseio.com',
    storageBucket: 'xxxx.appspot.com',
    projectId: 'xxxx',
});

const admin = require('firebase-admin');

admin.initializeApp({
    databaseURL: 'https://xxxx.firebaseio.com',
    credential: admin.credential.applicationDefault()
});
const db = admin.database();
const StateContext = require('./state-context');
const rcloadenv = require('@google-cloud/rcloadenv');

/**
 * This upsert json to Firebase realtime database.
 *
 * @param event
 * @returns {Promise}
 */
exports.upsertJson = (event) => {
    const gcs = require('@google-cloud/storage')();
    const os = require('os');
    const path = require('path');
    const file = event.data;
    const bucket = gcs.bucket(file.bucket);
    const fileNameWithDirectory = file.name;
    let stateContext = '';
    try {
        stateContext = new StateContext(fileNameWithDirectory);
    }
    catch (e){
        console.log(e + '. so do nothing.');
        return Promise.resolve();
    }
    const tempFilePath = path.join(os.tmpdir(), stateContext.getFileName());
    return bucket.file(fileNameWithDirectory).download({
        destination: tempFilePath
    }).then(() => {
        console.log('json file downloaded locally to', tempFilePath);
        const nReadLines = require('n-readlines');
        const liner = new nReadLines(tempFilePath);
        let updates = {};
        let line;
        while (line = liner.next()) {
            if (line.toString().trim().length > 0) {
                updates = stateContext.getState().getUpdateObject(line.toString(), updates);
            }
            else {
                console.info('skipped a blank line.');
            }
        }
        console.log('count', Object.keys(updates).length);
        return db.ref().update(updates);
    }).then(() => {
        console.log(tempFilePath, 'ended.');
        return Promise.resolve();
    }).catch((e) => {
        console.error(e);
        return Promise.resolve();
    });
};

やっていることとしては、CloudStorageのpathがmessages/*ならmessagesに、users/*ならusersにupsert（マージ）するだけ。
ファイルは一旦ダウンロードしてから登録している。
読み込みに関しては結構負荷がかかる。。。

state-context.js

'use strict';

const stateMap = new Map();
stateMap.set('messages', require('./message-state'));
stateMap.set('users', require('./user-state'));


/**
 * State context of change event in cloud storage.
 *
 * @param fileNameWithDirectory cloud storage fileNameWithDirectory
 * @constructor
 * @throws {error} invalid fileNameWithDirectory
 */
function StateContext(fileNameWithDirectory) {
    this.fileNameWithDirectory = fileNameWithDirectory;
    const fileNameArray = this.fileNameWithDirectory.split('/');
    this.fileName = fileNameArray[fileNameArray.length -1];
    this.directoryName = fileNameArray[0];
    if(!stateMap.has(fileNameArray[0])) {
        throw new Error(fileNameArray[0] + ' directory was edited.');
    }
    else {
        //do nothing.
    }
}

/**
 * This returns State object.
 */
StateContext.prototype.getState = function () {
    return stateMap.get(this.directoryName);
}

/**
 * This returns filename without directory.
 *
 * @returns {*}
 */
StateContext.prototype.getFileName = function () {
    return this.fileName;
}

module.exports = StateContext;

message-state.js

'use strict';
/**
 * get update objects for multiple update.
 *
 *
 * @param line
 * @param updates
 * @returns {*}
 */
exports.getUpdateObject = function(line, updates) {
    const jsonObject = JSON.parse(line);
    updates['/messages/' + jsonObject.messageId] = jsonObject;
    return updates;
}

user-state.js

'use strict';

/**
 * get update objects for multiple update.
 *
 * @param line
 * @param updates
 * @returns {*}
 */
exports.getUpdateObject = function(line, updates) {
    const jsonObject = JSON.parse(line);
    const params = {};
    params[jsonObject.messageId] = false;
    updates['/users/' + jsonObject.userId + '/' + jsonObject.messageId] = false;
    return updates;
}

#Deploy
##cloudbuild.yamlを使用してdeploy

cloudbuild.yaml

steps:
- name: 'gcr.io/cloud-builders/npm'
  args: ['prune']
  dir: 'functions'
- name: 'gcr.io/cloud-builders/npm'
  args: ['install']
  dir: 'functions'
- name: 'gcr.io/cloud-builders/npm'
  args: ['test']
  env: ['CI=1']
  dir: 'functions'
- name: 'gcr.io/cloud-builders/gcloud'
  args: ['beta', 'functions', 'deploy', 'upsertJson', '--entry-point', 'upsertJson', '--trigger-resource', 'storageバケット名', '--trigger-event', 'google.storage.object.finalize', '--memory', '1024MB', '--retry', '--timeout', '540s']
  dir: 'functions'

• unitテストについては後で書くが、avaを使用している。環境変数CIがないと失敗するので、設定。
• 一回のfunction起動で10000データさばくため、メモリは多め（要tuning）。
• 起動時間のリミットは最大限に。オフィシャルドキュメントに9分とあったので、540sで設定。
• Functionは必ず「成功」で終了するように作成した（ロギングはエラーだが）。Functionの起動が失敗になるパターンは主にtimeoutや、Function自体の起動の失敗等になる。retryオプションでretryするように設定することでtimeout時のretryを担保した。

Container Registryのビルドトリガーでgit pushを検知してcloudbuild.yamlを実行するよう設定。
以上でauto-deploy環境が整う。