Internet of things (IoT) is maturing rapidly and it is finding application across various industries. Every common device that we use is turning into the category of smart devices. Smart devices are basically IoT devices. These devices captures various parameters in and around their environment leading to generation of a huge amount of data. This data needs to be collected, processed, stored and analyzed in order to get actionable insights from them. To do so, we need to build data pipeline. In this blog we will be building a similar pipeline using Mosquitto, Kinesis, InfluxDB and Grafana. We will discuss all these individual components of the pipeline and the steps to build it.
Why the Analysis of IoT data is different
In an IoT setup, the data is generated by sensors that are distributed across various locations. In order to use the data generated by them we should first get them to a common location from where the various applications which want to process them can read it.
Network Protocol
IoT devices have low computational and network resources. Moreover, these devices write data in very short intervals thus high throughput is expected on the network. For transferring IoT data it is desirable to use lightweight network protocols. A protocol like HTTP uses a complex structure for communication resulting in consumption of more resources making it unsuitable for IoT data transfer. One of the lightweight protocol suitable for IoT data is MQTT which we are using in our pipeline. MQTT is designed for machine to machine (M2M) connectivity. It uses a publisher/subscriber communication model and helps clients to distribute telemetry data with very low network resource consumption. Along with IoT MQTT has been found to be useful in other fields as well.
Other similar protocols include Constrained Application Protocol (CoAP), Advanced Message Queuing Protocol (AMQP) etc.
Datastore
IoT devices generally collect telemetry about its environment usually through sensors. In most of the IoT scenarios, we try to analyze how things have changed over a period of time. Storing these data in a time series database makes our analysis simpler and better. InfluxDB is popular time series database which we will use in our pipeline. More about time series databases can be read here.
Pipeline Overview
The first thing we need for a data pipeline is data. As shown in the image above the data generated by various sensors are written to a topic in the MQTT message broker. To mimic sensors we will use a program which uses the MQTT client to write data to the MQTT broker.
The next component is Amazon Kinesis which is used for streaming data analysis. It closely resembles apache Kafka which is an open source tool used for similar purposes. Kinesis brings the data generated by a number of clients to a single location from where different consumers can pull it for processing. We are using Kinesis so that multiple consumers can read data from a single location. This approach scales well even if we have multiple message brokers.
Once the data is written to the MQTT broker a Kinesis producer subscribes to it and pull the data from it and writes it to the Kinesis stream, from the Kinesis stream the data is pulled by Kinesis consumers which processes the data and writes it to an InfluxDB which is a time series database.
Finally, we use Grafana which is a well-known tool for analytics and monitoring, we can connect it to many popular databases and perform analytics and monitoring. Another popular tool in this space is Kibana (the K of ELK stack)
Setting up a MQTT Message Broker Server:
For MQTT message broker we will use Mosquitto which is a popular open source message broker that implements MQTT. The details of downloading and installing mosquitto for various platforms are available here.
For Ubuntu, it can be installed using the following commands
sudo apt-add-repository ppa:mosquitto-dev/mosquitto-ppa
sudo apt-get update
sudo apt-get install mosquitto
service mosquitto statusSetting up InfluxDB and Grafana
The simplest way to set up both these components is to use their docker image directly
docker run --name influxdb -p 8083:8083 -p 8086:8086 influxdb:1.0
docker run --name grafana -p 3000:3000 --link influxdb grafana/grafana:3.1.1In InfluxDB we have mapped two ports, port 8086 is the HTTP API endpoint port while 8083 is the administration web server’s port. We need to create a database where we will write our data.
For creating a database we can directly go to the console at <influxdb-ip>:8083 and run the command: </influxdb-ip>
CREATE DATABASE "iotdata"Or we can do it via HTTP request :
curl -XPOST "http://localhost:8086/query" --data-urlencode "q=CREATE DATABASE iotdata”Creating a Kinesis stream
In Kinesis, we create streams where the Kinesis producers write the data coming from various sources and then the Kinesis consumers read the data from the stream. In the stream, the data is stored in various shards. For our purpose, one shard would be enough.
Creating the MQTT client
We will use the Golang client available in this repository to connect with our message broker server and write data to a specific topic. We will first create a new MQTT client. Here we can see the list of options we have for configuring our MQTT client.
Once we create the options object we can pass it to the NewClient() method which will return us the MQTT client. Now we can write data to the MQTT server. We have defined the structure of the data in the struct sensor data. Now to mimic two sensors which are writing telemetry data to the MQTT broker we have two goroutines which push data to the MQTT server every five seconds.
package publisher
import (
"config"
"encoding/json"
"fmt"
"log"
"math/rand"
"os"
"time"
"github.com/eclipse/paho.mqtt.golang"
)
type SensorData struct {
Id string `json:"id"`
Temperature float64 `json:"temperature"`
Humidity float64 `json:"humidity"`
Timestamp int64 `json:"timestamp"`
City string `json:"city"`
}
func StartMQTTPublisher() {
fmt.Println("MQTT publisher Started")
mqtt.DEBUG = log.New(os.Stdout, "", 0)
mqtt.ERROR = log.New(os.Stdout, "", 0)
opts := mqtt.NewClientOptions().AddBroker(config.GetMqttServerurl()).SetClientID("MqttPublisherClient")
opts.SetKeepAlive(2 * time.Second)
opts.SetPingTimeout(1 * time.Second)
c := mqtt.NewClient(opts)
if token := c.Connect(); token.Wait() && token.Error() != nil {
panic(token.Error())
}
go func() {
t := 20.04
h := 32.06
for i := 0; i < 100; i++ {
sensordata := SensorData{
Id: "CITIMUM",
Temperature: t,
Humidity: h,
Timestamp: time.Now().Unix(),
City: "Mumbai",
}
requestBody, err := json.Marshal(sensordata)
if err != nil {
fmt.Println(err)
}
token := c.Publish(config.GetMQTTTopicName(), 0, false, requestBody)
token.Wait()
if i < 50 {
t = t + 1*rand.Float64()
h = h + 1*rand.Float64()
} else {
t = t - 1*rand.Float64()
h = h - 1*rand.Float64()
}
time.Sleep(5 * time.Second)
}
}()
go func() {
t := 16.02
h := 24.04
for i := 0; i < 100; i++ {
sensordata := SensorData{
Id: "CITIPUN",
Temperature: t,
Humidity: h,
Timestamp: time.Now().Unix(),
City: "Pune",
}
requestBody, err := json.Marshal(sensordata)
if err != nil {
fmt.Println(err)
}
token := c.Publish(config.GetMQTTTopicName(), 0, false, requestBody)
token.Wait()
if i < 50 {
t = t + 1*rand.Float64()
h = h + 1*rand.Float64()
} else {
t = t - 1*rand.Float64()
h = h - 1*rand.Float64()
}
time.Sleep(5 * time.Second)
}
}()
time.Sleep(1000 * time.Second)
c.Disconnect(250)
}Create a Kinesis Producer
Now we will create a Kinesis producer which subscribes to the topic to which our MQTT client writes data and pull the data from the broker and pushes it to the Kinesis stream. Just like in the previous section here also we first create an MQTT client which connects to the message broker and subscribe to the topic to which our clients/publishers are going to write data to. In the client option, we have the option to define a function which will be called when data is written to this topic. We have created a function postDataTokinesisStream() which connects Kinesis using the Kinesis client and then writes data to the Kinesis stream, every time a data is pushed to the topic.
package producer
import (
"config"
"fmt"
"os"
"time"
"github.com/aws/aws-sdk-go/service/kinesis"
mqtt "github.com/eclipse/paho.mqtt.golang"
)
func postDataTokinesisStream(client mqtt.Client, message mqtt.Message) {
fmt.Printf("Received message on topic: %snMessage: %sn", message.Topic(), message.Payload())
streamName := config.GetKinesisStreamName()
kclient := config.GetKinesisClient()
var putRecordInput kinesis.PutRecordInput
partitionKey := message.Topic()
putRecordInput.PartitionKey = &partitionKey
putRecordInput.StreamName = &streamName
putRecordInput.Data = message.Payload()
putRecordOutput, err := kclient.PutRecord(&putRecordInput)
if err != nil {
fmt.Println(err)
} else {
fmt.Println(putRecordOutput)
}
}
func StartKinesisProducer() {
fmt.Println("Kinesis Producer Started")
c := make(chan os.Signal, 1)
opts := mqtt.NewClientOptions().AddBroker(config.GetMqttServerurl()).SetClientID("MqttSubscriberClient")
opts.SetKeepAlive(2 * time.Second)
opts.SetPingTimeout(1 * time.Second)
opts.OnConnect = func(c mqtt.Client) {
if token := c.Subscribe(config.GetMQTTTopicName(), 0, postDataTokinesisStream); token.Wait() && token.Error() != nil {
panic(token.Error())
}
}
client := mqtt.NewClient(opts)
if token := client.Connect(); token.Wait() && token.Error() != nil {
panic(token.Error())
} else {
fmt.Printf("Connected to %sn", config.GetMqttServerurl())
}
<-c
}Create a Kinesis Consumer
Now the data is available in our Kinesis stream we can pull it for processing. In the Kinesis consumer section, we create a Kinesis client just like we did in the previous section and then pull data from it. Here we first make a call to the DescribeStream method which returns us the shardId, we then use this shardId to get the ShardIterator and then finally we are able to fetch the records by passing the ShardIterator to GetRecords() method. GetRecords() also returns the NextShardIterator which we can use to continuously look for records in the shard until NextShardIterator becomes null.
package consumer
import (
"config"
"fmt"
"github.com/aws/aws-sdk-go/service/kinesis"
"velotio.com/dao"
)
func StartKinesisConsumer() {
fmt.Println("Kinesis Consumer Started")
client := config.GetKinesisClient()
streamName := config.GetKinesisStreamName()
var describeStreamInput kinesis.DescribeStreamInput
describeStreamInput.StreamName = &streamName
describeStreamOutput, err := client.DescribeStream(&describeStreamInput)
if err != nil {
fmt.Println(err)
} else {
fmt.Println(*describeStreamOutput.StreamDescription.Shards[0].ShardId)
}
var getShardIteratorInput kinesis.GetShardIteratorInput
getShardIteratorInput.ShardId = describeStreamOutput.StreamDescription.Shards[0].ShardId
getShardIteratorInput.StreamName = &streamName
shardIteratorType := "TRIM_HORIZON"
getShardIteratorInput.ShardIteratorType = &shardIteratorType
getShardIteratorOuput, err := client.GetShardIterator(&getShardIteratorInput)
if err != nil {
fmt.Println(err)
} else {
fmt.Println(*getShardIteratorOuput.ShardIterator)
}
var getRecordsInput kinesis.GetRecordsInput
getRecordsInput.ShardIterator = getShardIteratorOuput.ShardIterator
getRecordsOuput, err := client.GetRecords(&getRecordsInput)
//fmt.Println(getRecordsOuput)
if err != nil {
fmt.Println(err)
} else {
for *getRecordsOuput.NextShardIterator != "" {
i := 0
for i < len(getRecordsOuput.Records) {
//fmt.Println(len(getRecordsOuput.Records))
sdf := &dao.SensorDataFiltered{}
sdf.PostDataToInfluxDB(getRecordsOuput.Records[i].Data)
i++
}
getRecordsInput.ShardIterator = getRecordsOuput.NextShardIterator
getRecordsOuput, err = client.GetRecords(&getRecordsInput)
}
}
}Processing the data and writing it to InfluxDB
Now we do simple processing of filtering out data. The data that we got from the sensor is having fields sensorId, temperature, humidity, city, and timestamp but we are interested in only the values of temperature and humidity for a city so we have created a new structure ‘SensorDataFiltered’ which contains only the fields we need.
For every record that the Kinesis consumer receives it creates an instance of the SensorDataFiltered type and calls the PostDataToInfluxDB() method where the record received from the Kinesis stream is unmarshaled into the SensorDataFiltered type and send to InfluxDB. Here we need to provide the name of the database we created earlier to the variable dbName and the InfluxDB host and port values to dbHost and dbPort respectively.
In the InfluxDB request body, the first value that we provide is used as the measurement which is an InfluxDB struct to store similar data together. Then we have tags, we have used `city` as our tag so that we can filter the data based on them and then we have the actual values. For more details on InfluxDB data write format please refer here.
package dao
import (
"bytes"
"crypto/tls"
"encoding/json"
"fmt"
"net/http"
)
type SensorDataFiltered struct {
Temperature float64 `json:"temperature"`
Humidity float64 `json:"humidity"`
City string `json:"city"`
}
var dbName = "iotdata"
var dbHost = "184.73.62.30"
var dbPort = "8086"
func (sdf *SensorDataFiltered) PostDataToInfluxDB(Data []byte) {
err := json.Unmarshal(Data, &sdf)
if err != nil {
fmt.Println(err)
} else {
fmt.Println(sdf.Temperature, sdf.Humidity)
}
url := "http://" + dbHost + ":" + dbPort + "/write?db=" + dbName
humidity := fmt.Sprintf("%.2f", sdf.Humidity)
temperature := fmt.Sprintf("%.2f", sdf.Temperature)
city := sdf.City
requestBody := "sensordata,city=" + city + " humidity=" + humidity + ",temperature=" + temperature
req, err := http.NewRequest("POST", url, bytes.NewBuffer([]byte(requestBody)))
httpclient := &http.Client{
Transport: &http.Transport{
TLSClientConfig: &tls.Config{
InsecureSkipVerify: true,
},
},
}
resp, err := httpclient.Do(req)
if err != nil {
fmt.Println(err)
} else {
fmt.Println("Status code for influxdb data port request = ", resp.StatusCode)
}
defer resp.Body.Close()
}Once the data is written to InfluxDB we can see it in the web console by querying the measurement create in our database.
Putting everything together in our main function
Now we need to simply call the functions we discussed above and run our main program. Note that we have used `go` before the first two function call which makes them goroutines and they execute concurrently.
On running the code you will see the logs for all the stages of our pipeline getting written to the stdout and it very closely resembles real-life scenarios where data is written by IoT devices and gets processed in near real-time.
package main
import (
"time"
"velotio.com/consumer"
"velotio.com/producer"
"velotio.com/publisher"
)
func main() {
go producer.StartKinesisProducer()
go publisher.StartMQTTPublisher()
time.Sleep(5 * time.Second)
consumer.StartKinesisConsumer()
}Visualization through Grafana
We can access the Grafana web console at port 3000 of the machine on which it is running. First, we need to add our InfluxDB as a data source to it under the data sources option.
For creating dashboard go to the dashboard option and choose new. Once the dashboard is created we can start by adding a panel.
We need to add Influxdb data source that we added earlier as the panel data source and write queries as shown in the image below.
We can repeat the same process for adding another panel to the dashboard this time choosing a different city in our query.
Conclusion:
IoT data analytics is a fast evolving and interesting space. The number of IoT devices are growing rapidly. There is a great opportunity to get valuable insights from the huge amount of data generated by these device. In this blog, I tried to help you grab that opportunity by building a near real time data pipeline for IoT data. If you like it please share and subscribe to our blog.
