時系列DB InfluxDB2.0(MQTTを経由） メモ

InfluxDB(v2 Windows版）＋M5Stack UIFlow＋NodeRed（MQTT)で試したメモです。UIFlowでカスタムブロックをつくったり、ArduinoでならMQTTも不要で、もっと楽に使えそうです。

直接APIを使う版も書きました。

Node-REDのノードでInfluxDBに書き込む方法も書きました。(多分これが本命）

InfluxDB（時系列DB）

時系列DBはリレーショナルDBではない。リレーショナルDBに1億件ぐらいの多量データを入れると、めちゃくちゃ遅くなる。IoTでは、多量のデータを取り扱うのでリレーショナルDBは使いづらい。IoTのセンサーでは、データの更新はほぼなく、時系列で順々にデータが蓄積されて行っているので、検索とかでその特徴を生かされていると思う。

MQTTの準備

MQTTはnode-redで立ち上げます。
インストールはこれまでのページを見てください。

MQTT-Brokerを立ち上げるだけなので、これだけですね。

”デプロイ”を忘れない事。
PCのIPアドレスをcmdのipcondfig等で調べること。

M5Stack側の準備

UI-Flowはこんな感じ

MQTT　ServerのIPアドレスはPCのIPアドレスです。
MQTTのTopicをM5にしています。
MQTTに"room temp=xx.xx"と送信します。roomはTable名みたいなものになります。tempがフィールド名（カラム名）になります。詳しくは後半のLineInputを見てください。

Influxdbインストール方法

ダウンロード

からダウンロードして、展開して、C:\Program Files\InfluxData\influxdb\ に移動する（管理者権限が必要、フォルダを作成する必要あります。）下記のスクリプトをPowerShellで動かしても同じことができます。

wget https://dl.influxdata.com/influxdb/releases/influxdb2-2.0.8-windows-amd64.zip -UseBasicParsing -OutFile influxdb2-2.0.8-windows-amd64.zip
Expand-Archive .\influxdb2-2.0.8-windows-amd64.zip -DestinationPath 'C:\Program Files\InfluxData\'
mv 'C:\Program Files\InfluxData\influxdb2-2.0.8-windows-amd64' 'C:\Program Files\InfluxData\influxdb'

もう一つTelegrafもインストール
https://dl.influxdata.com/telegraf/releases/telegraf-1.20.2_windows_amd64.zip からダウンロードし、展開して　C:\Program Files\InfluxData\Telegraf\ に移動する。(ドキュメントのスクリプトが間違っているので、注意)
ディレクトリ構成はこんな感じになります。

c:\Program Files\InfluxData\influxdb\influx.exe
　　　　　　　　　　　　　　　  　　　　 \influxd.exe
                                     (他２つ)
                            \telegraf\telegraf.exe
                                      telegraf.conf

Influxdbの実行

実行

実行はPowerShell（管理者権限）で

cd -Path 'C:\Program Files\InfluxData\influxdb'
./influxd

設定

設定はGUIとCLIがあるらしい。ここではGUIで
ブラウザで localhost:8086を開いて、[Get Startted]をクリック

userとpassword、organization name、bucket name(DB名みたいなもの？）を登録

[Quick Start]をクリック

なんか動きました。

データは c:\user\ユーザーフォルダ\.infulxdbv2 に保存されていくみたいです。

Bucketの設定

[Data]の[Buckets]タブの初期設定で作ったbucketの枠内の[Settings]をクリック

1週間で消える設定にしてみました。こういうのが出来て楽ですよね。[Save Changes]をクリック

Teregraf

MQTTを受信できるように設定します。

設定

[Data]の[Sources]タブでMQTTを選択。多くの設定例が羅列しているので検索を使うと便利ですね。

ClipBoardにテンプレートをコピーし、teregraf.confをコピーしたものに書き換えます。自分はVisualSrudioCodeで編集しましたが好きなエディタを使えばよいと思います。
topicsを部分を書き換えて"M5"のTopicが受信できるようにします。

topics = [
    "M5",
  ]

[Data]の[teregraf]の[InfluxDB　Output　Plugin]　をクリック

クリップボードにコピーし　さっきの後ろに貼り付けます。

書込みにはTokenが必要なので作成します。
[Data]の[Tokens]タブの[+ Generate Taken]をクリック　Read/Writeを選択します。

書込みだけでOKなのでWrite Tokenを作ります。Write側の書き込むBucketsを選択し [Save]をクリックします。

生成されたTokenをコピーし、Tokenの部分を書き換えます。日付を参考に今作ったTokenの方を選びます。この画面で名前が変更できるのでわかりやすくしておくのも手ですね。名前の部分をクリックするとTokenが見れるのでクリップボードにコピーしましょう。

Tokenを書き換えて、こんな感じのteregraf.confになります。(tokenの部分は一部書き換えています）

[[inputs.mqtt_consumer]]
  ## Broker URLs for the MQTT server or cluster.  To connect to multiple
  ## clusters or standalone servers, use a seperate plugin instance.
  ##   example: servers = ["tcp://localhost:1883"]
  ##            servers = ["ssl://localhost:1883"]
  ##            servers = ["ws://localhost:1883"]
  servers = ["tcp://127.0.0.1:1883"]

  ## Topics that will be subscribed to.
  topics = [
    "M5",
  ]

  ## The message topic will be stored in a tag specified by this value.  If set
  ## to the empty string no topic tag will be created.
  # topic_tag = "topic"

  ## QoS policy for messages
  ##   0 = at most once
  ##   1 = at least once
  ##   2 = exactly once
  ##
  ## When using a QoS of 1 or 2, you should enable persistent_session to allow
  ## resuming unacknowledged messages.
  # qos = 0

  ## Connection timeout for initial connection in seconds
  # connection_timeout = "30s"

  ## Maximum messages to read from the broker that have not been written by an
  ## output.  For best throughput set based on the number of metrics within
  ## each message and the size of the output's metric_batch_size.
  ##
  ## For example, if each message from the queue contains 10 metrics and the
  ## output metric_batch_size is 1000, setting this to 100 will ensure that a
  ## full batch is collected and the write is triggered immediately without
  ## waiting until the next flush_interval.
  # max_undelivered_messages = 1000

  ## Persistent session disables clearing of the client session on connection.
  ## In order for this option to work you must also set client_id to identify
  ## the client.  To receive messages that arrived while the client is offline,
  ## also set the qos option to 1 or 2 and don't forget to also set the QoS when
  ## publishing.
  # persistent_session = false

  ## If unset, a random client ID will be generated.
  # client_id = ""

  ## Username and password to connect MQTT server.
  # username = "telegraf"
  # password = "metricsmetricsmetricsmetrics"

  ## Optional TLS Config
  # tls_ca = "/etc/telegraf/ca.pem"
  # tls_cert = "/etc/telegraf/cert.pem"
  # tls_key = "/etc/telegraf/key.pem"
  ## Use TLS but skip chain & host verification
  # insecure_skip_verify = false

  ## Data format to consume.
  ## Each data format has its own unique set of configuration options, read
  ## more about them here:
  ## https://github.com/influxdata/telegraf/blob/master/docs/DATA_FORMATS_INPUT.md
  data_format = "influx"
   [[outputs.influxdb_v2]]
  ## The URLs of the InfluxDB cluster nodes.
  ##
  ## Multiple URLs can be specified for a single cluster, only ONE of the
  ## urls will be written to each interval.
  ## urls exp: http://127.0.0.1:9999
  urls = ["http://localhost:8086"]

  ## Token for authentication.
  token = "V0-Jx4VPCtZ-IIPdqqz0r6ICy-nRxGN3qA4ZWRotVWXtPD8eBTRy1DOxVcXHBVA71DeiMLOhrFu1qT-ZOh******"

  ## Organization is the name of the organization you wish to write to; must exist.
  organization = "home"

  ## Destination bucket to write into.
  bucket = "Data1"

install&実行

teregrafのフォルダに移って、以下を実行し、動かします。

./telegraf --config telegraf.conf

確認

[Data Explorer]でデータを確認しましょう。
下の欄のデータを選択していき最後に[Submit]をクリックするとグラフが表示されます。

参考　InfluxDbのデータの書き込み

Line protocol

https://docs.influxdata.com/influxdb/cloud/reference/syntax/line-protocol/
こんな感じで書くらしいです。
measurementName fieldKey="field string value" 1556813561098000000

// Syntax

<measurement>[,<tag_key>=<tag_value>[,<tag_key>=<tag_value>]] <field_key>=<field_value>[,<field_key>=<field_value>] [<timestamp>]

// Example
myMeasurement,tag1=value1,tag2=value2 fieldKey="fieldValue" 1556813561098000000

tag（オプション）はIndexになるもので、Fieldはデータらしいです。Tagがあればそれでグループ化できるみたいです。
改行は newline character \n で表します。 Line protocol では、スペースに気を付けましょう。
Line protocolでは、tagとFieldでは改行**\n**は使えません。

measurementName,tagKey=tagValue fieldKey="fieldValue" 1465839830100400200
--------------- --------------- --------------------- -------------------
       |               |                  |                    |
  Measurement       Tag set           Field set            Timestamp

Measurement

(必須) measurement名. InfluxDBは一つのポイントごとに1つの measurement名を付けます。 Measurement名は大文字小文字を区別されます。measurement名は_で始めることはできません。
Data type: String

Tag set

(オプション) tag keyとtag valueの組み合わせになります。 Key-value との関係は**=** で表します。 複数のtag key-value の組み合わせは","で区切ります Tag keys と tag values は大文字小文字が区別されます。Tag keys は_で始めることはできません。

Key data type: String
Value data type: String

Field set

(必須) field keyとfield valueの組み合わせになります。最低1つは必要です。Field keys と string values は大文字小文字が区別されます。 Field keys は_で始めることはできません。

Key data type: String
Value data type: Float | Integer | UInteger | String | Boolean

**"**で始まるのは string field 値です。

measurementName fieldKey="field string value" 1556813561098000000

Timestamp

Optional – The unix timestamp for the data point. InfluxDB accepts one timestamp per point. もしtimestampがない場合は、InfluxDB はHostマシンのsystem time (UTC)を使います。
Data type: Unix timestamp
‎
タイムスタンプに関する注意
数値が観測された時刻 (InfluxDB によって受信されない) がデータ ポイントに含まれるようにするには、タイムスタンプを含めます。‎
‎タイムスタンプがナノ秒でない場合は、‎‎データをDBに書き込むときのタイムスタンプの精度を‎‎指定します。‎

空白（スペース）

空白はline protocolでは空白を使って値を分けます。 1つ目の空白はmeasurement名・the tag set をfield setと区別します。2つ目の空白はfield setとtimestampを区別します。

measurementName,tagKey=tagValue fieldKey="fieldValue" 1465839830100400200
                               |                     |
                           1st space             2nd space

データの形式

Type	方法	例
Float	IEEE-754 64-bit floating-point numbers. Default numerical type. InfluxDB supports scientific notation in float field values.	myMeasurement fieldKey=1.0 myMeasurement fieldKey=1 myMeasurement fieldKey=-1.234456e+78
Integer	語尾にiをつける -9223372036854775808i～ 9223372036854775807i	myMeasurement fieldKey=1i myMeasurement fieldKey=12485903i myMeasurement fieldKey=-12485903i
UInteger	語尾にuをつける 0u～18446744073709551615u	myMeasurement fieldKey=1u myMeasurement fieldKey=12485903u
String	ダブルコーテーションでくくる 長さは最大64kb	myMeasurement fieldKey="this is a string"
Boolean	以下の表現が使える True t, T, true, True, TRUE False f, F, false, False, FALSE	myMeasurement fieldKey=true myMeasurement fieldKey=false myMeasurement fieldKey=t myMeasurement fieldKey=f myMeasurement fieldKey=TRUE myMeasurement fieldKey=FALSE
Unix timestamp	Unix timestamp in a specified precision. Default precision is nanoseconds (ns). -9223372036854775806～ 9223372036854775806	myMeasurementName fieldKey="fieldValue" 1556813561098000000

引用符

Line protocolは**'と"**が使えます。

Element	Double quotes	Single quotes
Measurement	Limited *	Limited *
Tag key	Limited *	Limited *
Tag value	Limited *	Limited *
Field key	Limited *	Limited *
Field value	文字列のみ	なし
Timestamp	なし	なし

• ‎Line protocolは、計測名、タグキー、タグ値、およびフィールドキーで二重引用符と単一引用符を使用できますが、名前、キー、または値の一部として解釈されます。‎

特殊文字

Line protocolは特殊文字をstring elementsで使えます。**backslash ()**を使ってescapする必要があります。

Element	Escape characters
Measurement	Comma, Space
Tag key	Comma, Equals Sign, Space
Tag value	Comma, Equals Sign, Space
Field key	Comma, Equals Sign, Space
Field value	Double quote, Backslash

ほかの特殊文字はescape不要です。

Examples of special characters in line protocol

# Measurement name with spaces
my\ Measurement fieldKey="string value"

# Double quotes in a string field value
myMeasurement fieldKey="\"string\" within a string"

# Tag keys and values with spaces
myMeasurement,tag\ Key1=tag\ Value1,tag\ Key2=tag\ Value2 fieldKey=100

# Emojis
myMeasurement,tagKey=🍭 fieldKey="Launch 🚀" 1556813561098000000

backslashe\ のエスケープ
Line protocolは、バックスラッシュとエスケープ文字としてのバックスラッシュの両方をサポートします。連続する 2 つのバックスラッシュを使用すると、最初のバックスラッシュはエスケープ文字として解釈されます。例えば：

Backslashes	解釈
\	\
\	\
\\	\
\\	\
\\\	\\
\\\	\\

注釈行

Line protocolは行頭に#を使うとコメント行になります。

# This is a comment
myMeasurement fieldKey="string value" 1556813561098000000

名前付けの制限

Measurement names, tag keys, and field keys は最初にアンダーライン_は使えません。_ namespace は予約されています。

Duplicate points

‎ポイントは、計測名、タグセット、タイムスタンプによって一意に識別されます。同じ測定、タグセット、タイムスタンプを使用して行プロトコルを送信する場合、フィールドセットは古いフィールドセットと新しいフィールドセットの和集合となり、競合が新しいフィールドセットを優先します。‎

参考にしたところ