STEP-2 : InfluxDBでのBrokerからデータ受信確認
概要
Confluent Platform の「cp-all-in-one」 をベースにローカルのDockerコンテナ環境を構築し、IoTデータ生成Pythonプログラムから送信されるデータを Confluent でストリーミング処理し、InfluxDB＋Grafanaでリアルタイム可視化できることを確認しました。該当する Sink Connector を使用しています。
以下の絵(の記事) の環境をベースに、
Consumer側の可視化の構成を追加する形で確認します。
以下の３つのステップで上記内容を順次説明します。今回は STEP-2 について説明します。
STEP-1．Dockerコンテナ環境での Confluent Platform の追加構築
STEP-2．InfluxDBでのBrokerからのデータ受信確認
STEP-3．InfluxDB経由のGrafanaでのリアルタイムデータ可視化確認
ローカル環境
macOS Big Sur 11.3
python 3.8.3
Docker version 20.10.7, build f0df350 (CPUs:8, Memory:10GB, Swap:1GB)
influxdb の設定
- STEP-1 で作成したローカルDocker環境に対して、ローカルターミナルから「influxdb」に接続し情報を確認します。
$ docker exec -it influxdb /bin/bash
root@influxdb:/#
root@influxdb:/# uname -srn
Linux influxdb 5.10.25-linuxkit
root@influxdb:/#
root@influxdb:/# influxd version
InfluxDB v1.8.6 (git: 1.8 v1.8.6)
root@influxdb:/#
root@influxdb:/# influx -version
InfluxDB shell version: 1.8.6
root@influxdb:/#
root@influxdb:/# exit
exit
２. 「influxdb」にデータベース「IoTSample」を作成します。
$ docker exec -it influxdb influx
Connected to http://localhost:8086 version 1.8.6
InfluxDB shell version: 1.8.6
>
> create database IoTSample
>
> show databases
name: databases
name
----
_internal
IoTSample
>
コネクタの作成
- Confluent Platform の Control-Center へブラウザから http://localhost:9021 でアクセスします。
続いて、画面左側から「Connect」を選択し、新たに表示される「All Connect Clusters」画面から「connect-default」を選択し、その後に表示される「Connectors」画面右側から「+ Add connector」ボタンを押します。
新たに表示される「Browse」画面から「InfluxDBSinkConnector」を選択します。
その後に表示される「Add Connector」画面の「topics」項目で「topic_201」と「topic_202」を選択し、続く「name」項目に「InfluxDBSinkConnector_1」というコネクタ名を入力します。
コネクタ名を入力すると、入力項目が増えますので、以下の内容を入力し、最後に「Continu」ボタンを押します。
| カテゴリ | 入力項目 | 値 |
|---|---|---|
| Which topics do you want to get data from? | topics | topic_201, topic_202 |
| How should we connect to your data? | name | InfluxDBSinkConnector_1 |
| Common | tasks.max | 1 |
| Common | Value converter class | org.apache.kafka.connect.json.JsonConverter |
| InfluxDB | InfluxDB API URL* | http://influxdb:8086 |
| Write | InfluxDB Database | IoTSample |
| Write | Measurement Name Format | ${topic} |
| Additional Properties | value.converter.schemas.enable | false |
６. 設定内容に問題なければ、「Launch」ボタンを押します。
７. Connecotrs画面で作成したConnector「InfluxDBSinkConnector_1」が正常に起動していることを確認します。
InfluxDBでの確認
- ローカルターミナルから「influxdb」に接続し、データベース「IoTSample」にメジャーメント（テーブル）「topic_201」と「topic_202」が自動作成されていることを確認します。
$ docker exec -it influxdb influx
Connected to http://localhost:8086 version 1.8.6
InfluxDB shell version: 1.8.6
>
> use IoTSample
Using database IoTSample
>
> show measurements
name: measurements
name
----
topic_201
topic_202
>
２. ローカルコンピュータ上で、ここにある IoTデータ生成プログラムを実行します。50件のデータを生成します。
$ python IoTSampleData-v5.py --mode mq --count 50 --wait 1
３. 次に各メジャーメント（テーブル）にデータが Write されているか確認します。
>
> select * from topic_201
name: topic_201
time id iot_num iot_state proc section vol_1 vol_2
---- -- ------- --------- ---- ------- ----- -----
1626241925549000000 0 226-2461 島根県 111 M 198.1845397767455 57.268514136837176
1626241926554000000 1 145-2621 宮崎県 111 A 121.70584421224085 75.71752774135611
1626241927559000000 2 852-7187 熊本県 111 A 126.41080104674175 72.15478589353621
：
：
：
1626408886875000000 47 440-1826 兵庫県 111 J 116.80492255443481 58.16002504591993
1626408887881000000 48 238-2420 長野県 111 Q 168.35017898970486 68.57868248992546
1626408888886000000 49 175-8107 高知県 111 S 132.90782936265464 86.22280607722584
>
>
> select * from topic_202
name: topic_202
time IOT_NUM IOT_STATE PROC SECTION VOL_1 VOL_2 ZZTIME
---- ------- --------- ---- ------- ----- ----- ------
1626408840741000000 327-9411 埼玉県 111 C 127.4471407715012 71.60478903611397 2021-07-16T13:13:59.693142
1626408848776000000 910-2198 兵庫県 111 E 194.46476675784402 70.33439224608537 2021-07-16T13:13:59.693268
1626408863809000000 134-0246 北海道 111 E 123.3766530207224 61.78121028564689 2021-07-16T13:13:59.693468
1626408876870000000 214-7209 埼玉県 111 C 101.40696510178657 88.97291678787539 2021-07-16T13:13:59.693639
1626408878878000000 103-2715 福島県 111 C 160.6706712489061 53.167263544338155 2021-07-16T13:13:59.693665
>
これで正常にBrokerのトピック経由でInfluxDBでデータ受信できることを確認できました。
次のステップでは、このメジャーメント（テーブル）のデータを Grafana でリアルタイム可視化できることを確認してみます。
本課題のステップ情報
STEP-1．Dockerコンテナ環境での Confluent Platform の追加構築
STEP-2．InfluxDBでのBrokerからのデータ受信確認
STEP-3．InfluxDB経由のGrafanaでのリアルタイムデータ可視化確認