FirestoreのデータをElastcsearchへ移行してみた

FirebaseのFirestoreの検索がいまいちなので、
Elasticsearchを使うことにしてみました。
特にFirestoreのgeohashでの検索の精度がいまひとつ悪い、
というのが一番の理由です。

というわけで、FirestoreのデータをElasticsearchへインポートしてみました。

#Elasticsearchの準備
今回は、Elastic Cloudサービスを利用することにしました。
無料トライアルを利用してみて、このCloudサービスから利用した方がコントロールパネルが充実してそうなのでこちらを選択することにしました。

たぶん、料金はGCP、AWSから直で使っても同等の値段になると思います。

##indexの作成

testというindexを作成

##mapping設定

Elasticsearchでは基本データ投入時に自動で型を設定してくれるようですが、
それだと緯度経度がgeo_pointという型にならないので予めgeo_pointを定義します。

実行するとエラーに、

400 — Bad Request

{
  "status": 400,
  "error": {
    "root_cause": [
      {
        "reason": "Root mapping definition has unsupported parameters:  [items : {properties={location={type=geo_point}}}]",
        "type": "mapper_parsing_exception"
      }
    ],
    "type": "mapper_parsing_exception",
    "reason": "Root mapping definition has unsupported parameters:  [items : {properties={location={type=geo_point}}}]"
  }
}

前もって定義しようとしてもエラーになる、他にいい方法があるかもしれないけど仮のデータを投入してからgeo_pointを定義してみる。

###仮のデータ挿入

念のためデータができてるか確認

###改めてmapping設定

無事に成功、念のためmapping設定の確認

###仮のデータ削除

#データ移行処理

１件づつ移行するととても時間がかかるのでbulkという方法でまとめてINSERTします。
Functionsでも応用できるのでnodejsを利用します。

firebase_to_es.js

var admin = require("firebase-admin");
var serviceAccount = require("../project-0000000-firebase-adminsdk-xxxx-xxxxx.json");
admin.initializeApp({
  credential: admin.credential.cert(serviceAccount),
  databaseURL: "https://project-0000000.firebaseio.com"
});
var db = admin.firestore();
const esconfig = {
	"url" 		: "https://xxxxx.asia-northeast1.gcp.cloud.es.io:9243/",
	"username" 	: "elastic",
	"password" 	: "xxxxxxxxxxx"
};
const request = require('request-promise');

var startId = 0;

(async() => {

	do{
		items = await getItems(startId);

		if(items){
			var data = "";

			for(i in items){
				var item = items[i];
				item.location = {lat: item.location._latitude, lon: item.location._longitude};
				data += '{ "index":{ "_id" : "'+item.id+'" } }\n';
				data += JSON.stringify(item)+'\n'
			}


			let elasticsearchRequest = {
				method: "POST",
				uri: esconfig.url + "test/items/_bulk",
				auth: {
					username: esconfig.username,
					password: esconfig.password,
				},
				body: data,
				headers: {
					"Content-type": "application/x-ndjson",
				},		
				json: false
			};

			await request(elasticsearchRequest).then(response => {
				//console.log('Elasticsearch response', response);
			})		
		}

	}while(items)


	console.log("finish!");


})();

async function getItems(id){
	console.log("id", id);

	var snapshot = await new Promise( (resolve, reject) => {
		db.collection('items').where('id', '>', id).orderBy('id').limit(50).get()
				.then(snapshot => {
					if (snapshot.empty) {
						console.log('No matching documents.');
						resolve(null);
						return;
					}
					resolve(snapshot);
					return;
				})
				.catch(err => {
					console.log('Error getting documents', err);
					resolve(null);
				});
	}).catch(function(error) {

		console.log("The error is handled, continue normally");

	});

	if(snapshot===null){
		return false;
	}

	var items = [];

	snapshot.forEach(doc => {
		items.push(doc.data());
		startId = parseInt(doc.id);
	});	

	return items;

}

移行処理の実行

$ node firebase_to_es.js

##KibanaでデータをMapで確認

お値段が↓こんな感じで、リーズナブルで運用できそうです。