2020-04-04

【InfluxDB】データの重複(上書き)の避け方 + bitFlyerの板データを効率良く保存

前回までの調査でbitFlyerから送られてくる板データをそのままDBに入れるのは効率が悪いと考えられる。より効率をよくするためにはどうすればいいのだろうか。

前回から分かったこと

データは配列で送られてくる
DBに配列は入れられない
データの個数は280件/秒

★10分程度スナップショット＋差分の件数を自前で計測：約280件/秒の平均と出た。データが送られてくる回数や通信量は今回問題ではないので内部の { price, size } で1件とカウントしている。

以上から

データを配列で入れるのは非効率なため1件ごとに分割して記録する必要があると考える。

この場合データ登録にDBが追いつけるのかを調査する。

秒間270レコードは多いのか？

昨日記事を書いた後に公式ドキュメントを精査していたところHardware sizing guidelinesというページがあり性能に対してどれぐらいのデータが記録できるのかの目安とできそうだった。

まずは次の表を参考にする。

Hardware sizing guidelines

一番上の最小構成を見てほしい。Writes per secondは5000未満となっており、今回私のデータが余裕であることが分かる（1件あたりのデータ量も少ない）

では、最小構成はどんなPCだろうか。

家庭で常時稼働させられるLinux PCといえば現時点ではRaspberry Pi 4が最適だろう。

Raspberry Pi 4のスペックはCPU4コア、メモリ4Gで最低スペックをクリアしている。IOPSに関して言えばSSDが安物でも数万と言われているので余裕である。

しかし気になる項目がある。Unique series < 100,000 とはなんだろうか。

公式のドキュメント( Seriesの説明 )を読む必要がある。

A logical grouping of data defined by shared measurement, tag set, and field key.

この説明では短すぎてSeriesとは何かが分かりにくいが、InfluxDB key conceptsというページにもう少し詳しく具体例が乗っていた。

TagとFieldの違いについてはここでは説明しないがTagにはよくアクセスするデータを入れて検索を高速化させる役割があるためこのあたりのコストは注意が必要だ。

例えば今回のデータ { price:値, size:値 } で考えればFieldが2個なのでUnique seriesは「２」で良いはずだが、ここにtag=1 の { price:値, size:値 } tag=2の { price:値, size:値 }とすると倍になり「４」になる。

つまりtagの数だけ指数関数的に増えると考えている。

結論：いずれにせよ今回のパターンではデータの数自体は多いものの項目は極めて限定的であり「Raspberry Pi 4 + SSD」のようなマシンでも十分データをストアできると考える。

余談：次のスキーマは危険

以前の謎のチャート製作時に使用したJSON形式をそのままInfluxDBに入れる場合は危険だ。

謎のチャートのデータ構造は { price:値, size:値 } を変換し { 価格(キー）:サイズ(値) } にしている。

例えば10万円のBTCに0.5の指値が入っていた場合は { 100000 : 0.5 } というデータになる構造だ。

この場合データ量としてはスマートだがInfluxDBの指標に換算するとFieldsの数＝∞（100万幅は軽い）ため簡単にUnique seriesの目安を超えてしまう可能性がある。

注意しておきたい。

実際にデータを書き込んでみる

まずは送られてきたデータの配列構造を分割しInfluxDB用に変換する。

スキーマの定義

    schema: [
        {
            measurement: 'lightning_board_FX_BTC_JPY', //差分もスナップショットも一つに纏める
            fields: {
                price: Influx.FieldType.INTEGER,
                size: Influx.FieldType.FLOAT,
            },
            tags: [
                'type', //差分とスナップショットはTagを使って分ける
                'bid_or_ask' //bidかaskか
                'uniq' //重要：後述
            ]
        }
    ]

今回はTagsを使って差分かスナップショットかどうか(type)、bidかaskか(bid_or_ask)を明示しておく。これによりクエリを発行するときにパフォーマンスが向上すると思われる。

uniqについては後述（重要）

余談：tagsはstring:stringのobject限定なのでFieldTypeを設定できないらしく配列にキー名を入れるのみ（自動的にStringと認識)

データ登録部

データ登録は「バッチ(batch)」という概念で行う。

これは{ price:値, size:値 }を1件として、複数同時にレコード登録する方法だ。

writePoints()関数にオブジェクトを配列として送る事で一度に複数のレコードが記録される。

//スナップショットを登録するコード
//socket.ioにて接続
socket_bF_client.on("lightning_board_snapshot_FX_BTC_JPY", (message) => {
    const asks_data = message.asks.map((asks, index) => {
        return {
            measurement: 'lightning_board_FX_BTC_JPY',
            tags: {
                type: 'snapshot',
                bid_or_ask: 'ask',
                uniq: index
            },
            fields: {
                price: asks.price,
                size: asks.size
            }
        }
    })

    const bids_data = message.bids.map((bids, index) => {
        return {
            measurement: 'lightning_board_FX_BTC_JPY',
            tags: {
                type: 'snapshot',
                bid_or_ask: 'bid',
                uniq: index
            },
            fields: {
                price: bids.price,
                size: bids.size
            }
        }
    })

    influx.writePoints([...asks_data, ...bids_data]).then(() => {
        console.log("ok")
    }).catch(err => {
        console.error(`Error saving data to InfluxDB! ${err.stack}`)
    })
})