1. はじめに

IoTセンサノードは作ったけど、データの蓄積方法について悩んでいるかた向けの内容です。ラズパイなど、サーバとなるホストにnode-redとinfluxDBがインストール済みであることを前提にしています(あと、node-redへのinfluxDBノードのインストールも)。インストールは簡単なので、まだの場合は検索してみてください。
なお、influxDBのバージョンは1.6以前を想定しています。v1.8やv2.0以降は少し異なるので、この内容では動かないかもしれません。

2. データの名前などの検討

データを蓄積する前に検討する必要があるのが、データベース名、項目名、分類名です。もし、データベース名など、それぞれの名称が、予め用意されているものがあればそれでも良いです。

何のデータを取得したのか（羅列する）
それはどこのデータなのか
どんなデータなのか

などを考えます。
例えば次のようにします。センサノードが、部屋Aの天井と床の温度を測っているというものです。データベース名は適当にiotdataにします。

データベース名：iotdata
分類名(measurement名) : RoomA
項目名 :tempRoomCeil , tempRoomFloor

influxDBにおける、データベース名、分類名(Measurement名)、項目名(field)の概念を図にすると次のようになります。

これを表計算ソフトに置き換えると次のような概念です。

つまり、これらの名前を決める作業と考えて良いでしょう。

3. influxDBでデータベース名の設定

データベースの作成はとても簡単なのですが、ターミナルで文字を入力する必要があります。
ターミナルを立ち上げて、次を入力します。

influx

その後、InfluxDB shell version: 1.6とか、色々とメッセージが表示され、最後に >という文字と四角(カーソル）が点滅している状態になります。この状態で次を入力します。データベース名をiotdataとします。

create iotdata

と入力したら、次でinfluxdbのシェルモードを終了します。

exit

途中でエラーが出たら、文字が間違っていないか確認してください。ただし、データベース名は、間違ってもエラーは出ませんので、きちんと予定していた名称が入力されているか確認しましょう。

4. node-redでのプログラム

4.1 概要

作るのは次のプログラムです。

前提としては次の通りです。

センサノードからMQTTで受信
センサノードから2つの数字をカンマ区切りで送られている（例： 20.1 , 20,8 ）

MQTTノードの置き方は割愛します。
MQTTノードの後ろに、csvノード、functionノード、最後に influxdb outノードと置きます。
csvノードは特に設定は不要です。

4.2 functionノードの内容

センサノードが出力している順序が天井、床、とします（この順序が違うと違う名前をつけることになってしまうので要注意）。
csvノードが出力したオブジェクトは、msg.payload.col1とmsg.payload.col2になっています。このノードでは、それを、新たに名前を付けて、保存しなおします。（つまり、col1という名称をtempRoomCeilにして、col2という名称をtempRoomFloorに変更したわけです）
この項目名が、influxDB側でそのままフィールド名として反映されます。
なお、csvノードの出力は文字列のままなので、Numberで数値に変更します。

var tempRoomCeil = Number(msg.payload.col1);
var tempRoomFloor = Number(msg.payload.col2);
msg.payload={tempRoomCeil:tempRoomCeil,tempRoomFloor:tempRoomFloor};
return msg;

4.3 influxDB outノードの設定

influxdb outノードをダブルクリックします。serverの欄の脇にある鉛筆のアイコンをクリックします。

host名は127.0.0.1と入力されているはずです。あとは、データベース名に、決めた名称のiotdataと入力し、追加を押します。

Measurement名に、決めた名称のRoomAと入力し、完了を押します。

4.4 コード

完成したノードはこちらです。node-redの三のようなアイコンをクリックして、「読み込み」を選択し、出てきた画面にペーストすると、ノードが置けます。

nodered_csv2obj_influx.json

[
    {
        "id": "a6d91f91.e012b",
        "type": "mqtt in",
        "z": "65c8cc70.842004",
        "name": "",
        "topic": "myTopic",
        "qos": "2",
        "datatype": "utf8",
        "broker": "7e153fb7.ee0ad8",
        "x": 120,
        "y": 180,
        "wires": [
            [
                "89603155.dda7b8"
            ]
        ]
    },
    {
        "id": "89603155.dda7b8",
        "type": "csv",
        "z": "65c8cc70.842004",
        "name": "",
        "sep": ",",
        "hdrin": "",
        "hdrout": "none",
        "multi": "one",
        "ret": "\\n",
        "temp": "",
        "skip": "0",
        "strings": true,
        "include_empty_strings": "",
        "include_null_values": "",
        "x": 270,
        "y": 180,
        "wires": [
            [
                "672ebe47.2e04f8"
            ]
        ]
    },
    {
        "id": "672ebe47.2e04f8",
        "type": "function",
        "z": "65c8cc70.842004",
        "name": "",
        "func": "var tempRoomCeil = Number(msg.payload.col1);\nvar tempRoomFloor = Number(msg.payload.col2);\nmsg.payload={tempRoomCeil:tempRoomCeil,tempRoomFloor:tempRoomFloor};\nreturn msg;",
        "outputs": 1,
        "noerr": 0,
        "initialize": "",
        "finalize": "",
        "x": 420,
        "y": 180,
        "wires": [
            [
                "c6930ba4.d98cf8"
            ]
        ]
    },
    {
        "id": "c6930ba4.d98cf8",
        "type": "influxdb out",
        "z": "65c8cc70.842004",
        "influxdb": "cbd5b535.0b89e8",
        "name": "",
        "measurement": "RoomA",
        "precision": "",
        "retentionPolicy": "",
        "x": 650,
        "y": 180,
        "wires": []
    },
    {
        "id": "7e153fb7.ee0ad8",
        "type": "mqtt-broker",
        "z": "",
        "name": "",
        "broker": "localhost",
        "port": "1883",
        "clientid": "",
        "usetls": false,
        "compatmode": false,
        "keepalive": "60",
        "cleansession": true,
        "birthTopic": "",
        "birthQos": "0",
        "birthPayload": "",
        "closeTopic": "",
        "closeQos": "0",
        "closePayload": "",
        "willTopic": "",
        "willQos": "0",
        "willPayload": ""
    },
    {
        "id": "cbd5b535.0b89e8",
        "type": "influxdb",
        "z": "",
        "hostname": "127.0.0.1",
        "port": "8086",
        "protocol": "http",
        "database": "iotdata",
        "name": "",
        "usetls": false,
        "tls": ""
    }
]

5. データの取り出し

一定期間データが溜まったら、そのデータをcsvファイルに出力してみましょう。
ターミナルから次のように入力します。

influx -database 'iotdata' -execute 'SELECT * FROM RoomA' -format csv > iotdata.csv

間違いがあれば、エラーが出ます。正しければ、ユーザのフォルダに、iotdata.csvが出来上がります。
これを表計算ソフトで読み込ませたりすることができます。
ラズパイであれば、USBメモリを挿して、コピーしてPCにデータを移すと良いでしょう。

6. その他

6.1 デバッグ

短い内容なので、不要と思いますが、センサが送るであろう対向処理もnode-red側で準備して、デバッグできるようにしましょう。

nodered_csv2obj_influx_debug.json

[
    {
        "id": "ed941a71.05273",
        "type": "inject",
        "z": "65c8cc70.842004",
        "name": "",
        "props": [
            {
                "p": "payload"
            },
            {
                "p": "topic",
                "vt": "str"
            }
        ],
        "repeat": "",
        "crontab": "",
        "once": false,
        "onceDelay": 0.1,
        "topic": "",
        "payload": "",
        "payloadType": "date",
        "x": 160,
        "y": 80,
        "wires": [
            [
                "a81631e8.80424"
            ]
        ]
    },
    {
        "id": "a81631e8.80424",
        "type": "function",
        "z": "65c8cc70.842004",
        "name": "",
        "func": "var strA=(String)(Math.random()*1+20);\nvar strB=(String)(Math.random()*1+20);\nvar strOut=strA+\" , \"+strB;\nmsg.payload=strOut;\nreturn msg;\n",
        "outputs": 1,
        "noerr": 0,
        "initialize": "",
        "finalize": "",
        "x": 350,
        "y": 80,
        "wires": [
            [
                "3fd8b6f3.f0d032"
            ]
        ]
    },
    {
        "id": "3fd8b6f3.f0d032",
        "type": "mqtt out",
        "z": "65c8cc70.842004",
        "name": "",
        "topic": "myTopic",
        "qos": "",
        "retain": "",
        "broker": "7e153fb7.ee0ad8",
        "x": 550,
        "y": 80,
        "wires": []
    },
    {
        "id": "a6d91f91.e012b",
        "type": "mqtt in",
        "z": "65c8cc70.842004",
        "name": "",
        "topic": "myTopic",
        "qos": "2",
        "datatype": "utf8",
        "broker": "7e153fb7.ee0ad8",
        "x": 120,
        "y": 140,
        "wires": [
            [
                "89603155.dda7b8"
            ]
        ]
    },
    {
        "id": "89603155.dda7b8",
        "type": "csv",
        "z": "65c8cc70.842004",
        "name": "",
        "sep": ",",
        "hdrin": "",
        "hdrout": "none",
        "multi": "one",
        "ret": "\\n",
        "temp": "",
        "skip": "0",
        "strings": true,
        "include_empty_strings": "",
        "include_null_values": "",
        "x": 270,
        "y": 140,
        "wires": [
            [
                "672ebe47.2e04f8"
            ]
        ]
    },
    {
        "id": "8be8e886.52f7c8",
        "type": "debug",
        "z": "65c8cc70.842004",
        "name": "",
        "active": true,
        "tosidebar": true,
        "console": false,
        "tostatus": false,
        "complete": "false",
        "statusVal": "",
        "statusType": "auto",
        "x": 650,
        "y": 280,
        "wires": []
    },
    {
        "id": "672ebe47.2e04f8",
        "type": "function",
        "z": "65c8cc70.842004",
        "name": "",
        "func": "var tempRoomCeil = Number(msg.payload.col1);\nvar tempRoomFloor = Number(msg.payload.col2);\nmsg.payload={tempRoomCeil:tempRoomCeil,tempRoomFloor:tempRoomFloor};\nreturn msg;",
        "outputs": 1,
        "noerr": 0,
        "initialize": "",
        "finalize": "",
        "x": 420,
        "y": 140,
        "wires": [
            [
                "c6930ba4.d98cf8",
                "8be8e886.52f7c8"
            ]
        ]
    },
    {
        "id": "c6930ba4.d98cf8",
        "type": "influxdb out",
        "z": "65c8cc70.842004",
        "influxdb": "cbd5b535.0b89e8",
        "name": "",
        "measurement": "RoomA",
        "precision": "",
        "retentionPolicy": "",
        "x": 650,
        "y": 140,
        "wires": []
    },
    {
        "id": "2c0edbfb.744e1c",
        "type": "influxdb in",
        "z": "65c8cc70.842004",
        "influxdb": "cbd5b535.0b89e8",
        "name": "",
        "query": "select * from RoomA",
        "rawOutput": false,
        "precision": "",
        "retentionPolicy": "",
        "x": 470,
        "y": 200,
        "wires": [
            [
                "a23ebf11.a8369",
                "8be8e886.52f7c8"
            ]
        ]
    },
    {
        "id": "91618ac1.8be3b",
        "type": "inject",
        "z": "65c8cc70.842004",
        "name": "",
        "props": [
            {
                "p": "payload"
            },
            {
                "p": "topic",
                "vt": "str"
            }
        ],
        "repeat": "",
        "crontab": "",
        "once": false,
        "onceDelay": 0.1,
        "topic": "",
        "payload": "",
        "payloadType": "date",
        "x": 170,
        "y": 200,
        "wires": [
            [
                "2c0edbfb.744e1c"
            ]
        ]
    },
    {
        "id": "a23ebf11.a8369",
        "type": "csv",
        "z": "65c8cc70.842004",
        "name": "",
        "sep": ",",
        "hdrin": "",
        "hdrout": "none",
        "multi": "one",
        "ret": "\\n",
        "temp": "",
        "skip": "0",
        "strings": true,
        "include_empty_strings": "",
        "include_null_values": "",
        "x": 450,
        "y": 280,
        "wires": [
            [
                "8be8e886.52f7c8"
            ]
        ]
    },
    {
        "id": "7e153fb7.ee0ad8",
        "type": "mqtt-broker",
        "z": "",
        "name": "",
        "broker": "localhost",
        "port": "1883",
        "clientid": "",
        "usetls": false,
        "compatmode": false,
        "keepalive": "60",
        "cleansession": true,
        "birthTopic": "",
        "birthQos": "0",
        "birthPayload": "",
        "closeTopic": "",
        "closeQos": "0",
        "closePayload": "",
        "willTopic": "",
        "willQos": "0",
        "willPayload": ""
    },
    {
        "id": "cbd5b535.0b89e8",
        "type": "influxdb",
        "z": "",
        "hostname": "127.0.0.1",
        "port": "8086",
        "protocol": "http",
        "database": "iotdata",
        "name": "",
        "usetls": false,
        "tls": ""
    }
]

6.2 influxDB側の確認

influxDBのデータベースに保存されているか、確認したい場合、2つの方法があります。

influxのコマンドで確かめる
node-redで確かめる。

6.2.1 influxのコマンド

influxDB v1系は、SQLライクなコマンドを使うことができます。

ターミナルから、influxと入力します。use データベース名と入力し、select * from メジャメント名で、内容表示されます。

6.2.2 node-redのinfluxDBノード

influxDB inノードを使います。上のフローには含まれていますが、Queryの部分に、select * from メジャメント名を入力します。

この出力をそのまま観ることもできますが、csvノードで、csvにすることもできます。ただし、時刻の項目だけは省略されます。arrayのobjectという部分をクリックすると内容が表示されます。これで、influxには正しく保存されていることが判ります。

6.3 項目名を付ける処理について

今回の説明は、センサノードから、CSV形式でデータを送られた場合としました。しかし、センサノード自身がデータの項目名をつけるべきという考え方もあります。その場合は、センサーノード自身が、JSON形式で送るべきでしょう。その場合、MQTTノードで、受信をMQTTとして、influxdb-outノードをそのまま接続するだけで済みます。

7.おわりに

今回は、node-redとinfluxdbでデータ蓄積するという内容でした。influxdb-outノードはとても便利で、field名を渡された項目名にしてくれます。ただ、influxdbが変化しつづけており、この方法もいつまで使えるか判りません。とにかくinfluxdbを使おうという方に、参考になればと思います。