[Apache Hop] ローコードでMQTT送受信してみた

# 紹介したいこと
- ローコードでMQTTを送受信する方法を紹介します。
- HopにはMQTTをPublish, Subscribeする部品がある。
- HopでのETLの定義の雰囲気を伝えたい for Beginner。

# 対象者
- Apache Hop (データ統合プラットフォーム) 、ETLを試してみたい方。
- ローコードでMQTTを送受信してETL処理に連携させたい方。

# 前提
- Dockerを動かすことができる(MQTT Broker)。
- Apache Hopをインストールしている、HOP GUIを操作できる、簡単な変換処理を定義できる。
- MQTT部品は標準部品に含まれてない。EXTERNAL PLUGINSのため、以下のサイトの手順沿ってビルドして配置する必要がある。

https://hop.apache.org/manual/latest/plugins/external-plugins.html#_transforms

# 全体構成
MQTTの送受信の動きを確認するため、以下の全体構成、データの流れを考えて、試してみた。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/2648069/9bfd318f-fa34-c63f-9509-997aca5ed734.png)
MQTTとは
https://kfep.jp/solution/iot-mqtt/mqtt

# 手順の流れ
- MQTT Brokerを起動する
- MQTTをPublishするPipelineを作成する
- MQTTをSubscribeするPipelineを作成する
- 動作確認する

## MQTT Brokerを起動する
以下のDocker Composeを定義して、docker-compose up -dでBrokerを起動する。
mosquittoをMQTTサーバとして使う。
```
version: "3"
services:

  broker:
    image: eclipse-mosquitto
    volumes:
      - ./mosquitto/config:/mosquitto/config
      - ./mosquitto/data:/mosquitto/data
      - ./mosquitto/log:/mosquitto/log
    user: "1000:1000"
    ports:
      - 1883:1883
```
## MQTTをPublishするPipelineを作成する
以下のようなパイプラインを作成して実行する。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/2648069/8e8792d0-61bc-337b-a8c2-904057015b4f.png)

- Generate rows
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/2648069/f2ecea66-749b-a443-b524-6d1d3fa56546.png)
Never stop generating rows: チェックをいれる
Interbal in ms: 5000 5秒周期で行を生成しつづける

- Add sequence
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/2648069/70659856-29fe-cbce-7262-853e485bf29f.png)
Name of value: valuename → このフィールド名でデータがながれる
Start at Value: 1 : 1から開始
Increment by: 1 : 1ずつインクリメント
Maximum value: 100 : 1~100までインクリメントでまた1にもどる

- MQTT Publisher
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/2648069/5c10bdc7-c3aa-bc00-10f0-516bcb948fd3.png)
Broker URL: 環境にあわせて設定すること。
Credentials, SSLは、環境にあわせて設定すること。
Incoming message filed: valuename: 流れてきたデータをmessageとしてPublishする

## MQTTをSubscribeするPipelineを作成する
以下のようなパイプラインを作成して実行する。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/2648069/81396270-c389-8b26-671a-4a14e81623b4.png)

- MQTT Subscriber
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/2648069/ed52ad70-c42d-fda8-ee60-1493e66bfebf.png)
Broker URL: 環境にあわせて設定すること。
Credentials, SSLは、環境にあわせて設定すること。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/2648069/4e3a5579-d243-6ff6-fa7a-8eb11da9c200.png)
受信するトピックスの名前、データ型を定義する。

## 動作確認する

Publishする側は、以下のインクリメントしたデータを送信し続ける。定周期 5秒ごとに。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/2648069/93d879bb-ba9a-fe21-580e-286710176a5a.png)

送信されたデータは、 MQTT Brokerを介してSubscriberに配信される。

Subscribeする側は、以下のようにデータが受信される。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/2648069/25bbce7c-ba75-a9f0-0b52-857e19c2fed9.png)

# まとめ
- このように**MQTT部品を利用して、MQTTのデータをPublisher, Subscribeできる**ことを確認できた。IoTデータなどがMQTT経由でApacheHopでデータを受信できそう。
- ApacheHopには、mongoDB、PostgerSQL、neo4j, splunkとの連携する機能がある。MQTTで受信したデータを変換して保存することができるかもしれない。
- Kafka部品もあるので、MQTTのデータをKafkaに連携させることもできそう。MQTTのデータをKSQLなどで集計することもできるかもしれない。

# 注意事項
- データ量が増加したり、データの送信頻度が増えたときに、どのくらいスループットが保証できるかは、要件にあわせて試してみてる必要があると思う。
- 1Pipelineにどのくらい定義できるのか？何個最大でうごかせるか？などはマシンスペック、ネットワークなど環境によりそう。
- MQTT部品でセキュリティの設定したときの動作は未確認。それも要件によって確認が必要だと思う。

# 参考

Streaming Apache Hop and MQTT
https://diethardsteiner.github.io/hop/2021/02/08/Apache-Hop-An-introduction-to-MQTT.html