In the following post, we will explore the integration of several open-source software applications to build an IoT edge analytics stack, designed to operate on ARM-based edge nodes. We will use the stack to collect, analyze, and visualize IoT data without first shipping the data to the Cloud or other external systems.

The Edge

Edge computing is a fast-growing technology trend, which involves pushing compute capabilities to the edge. Wikipedia describes edge computing as a distributed computing paradigm that brings computation and data storage closer to the location needed to improve response times and save bandwidth. The term edge commonly refers to a compute node at the edge of a network (edge device), sitting in close proximity to the source a data and between that data source and external system such as the Cloud.

In his recent post, 3 Advantages (And 1 Disadvantage) Of Edge Computing, well-known futurist Bernard Marr argues reduced bandwidth requirements, reduced latency, and enhanced security and privacy as three primary advantages of edge computing. Due to techniques like data downsampling, Marr advises one potential disadvantage of edge computing is that important data could end up being overlooked and discarded in the quest to save bandwidth and reduce latency.

David Ricketts, Head of Marketing at Quiss Technology PLC, estimates in his post, Cloud and Edge Computing — The Stats You Need to Know for 2018, the global edge computing market is expected to reach USD 6.72 billion by 2022 at a compound annual growth rate of a whopping 35.4 percent. Realizing the market potential, many major Cloud providers, edge device manufacturers, and integrators are rapidly expanding their edge compute capabilities. AWS, for example, currently offers more than a dozen services in the edge computing category.

Internet of Things

Edge computing is frequently associated with the Internet of Things (IoT). IoT devices, industrial equipment, and sensors generate data, which is transmitted to other internal and external systems, often by way of edge nodes, such as an IoT Gateway. IoT devices typically generate time-series data. According to Wikipedia, a time series is a set of data points indexed in time order — a sequence taken at successive equally spaced points in time. IoT devices typically generate continuous high-volume streams of time-series data, often on the scale of millions of data points per second. IoT data characteristics require IoT platforms to minimally support temporal accuracy, high-volume ingestion and processing, efficient data compression and downsampling, and real-time querying capabilities.

The IoT devices and the edge devices, such as IoT Gateways, which aggregate and transmit IoT data from these devices to external systems, are generally lower-powered, with limited processor, memory, and storage capabilities. Accordingly, IoT platforms must satisfy all the requirements of IoT data while simultaneously supporting resource-constrained environments.

IoT Analytics at the Edge

Leading Cloud providers AWS, Azure, Google Cloud, IBM Cloud, Oracle Cloud, and Alibaba Cloud all offer IoT services. Many offer IoT services with edge computing capabilities. AWS offers AWS IoT Greengrass. Greengrass provides local compute, messaging, data management, sync, and ML inference capabilities to edge devices. Azure offers Azure IoT Edge. Azure IoT Edge provides the ability to run AI, Azure and third-party services, and custom business logic on edge devices using standard containers. Google Cloud offers Edge TPU. Edge TPU (Tensor Processing Unit) is Google’s purpose-built application-specific integrated circuit (ASIC), designed to run AI at the edge.

IoT Analytics

Many Cloud providers also offer IoT analytics as part of their suite of IoT services, although not at the edge. AWS offers AWS IoT Analytics, while Azure has Azure Time Series Insights. Google provides IoT analytics, indirectly, through downstream analytic systems and ad hoc analysis using Google BigQuery or advanced analytics and machine learning with Cloud Machine Learning Engine. These services generally all require data to be transmitted to the Cloud for analytics.

The ability to analyze IoT data at the edge, as data is streamed in real-time, is critical to a rapid feedback loop. IoT edge analytics can accelerate anomaly detection, improve predictive maintenance capabilities, and expedite proactive inventory replenishment.

The IoT Edge Analytics Stack

In my opinion, the ideal IoT edge analytics stack is comprised of lightweight, purpose-built, easily deployable and manageable, platform- and programming language-agnostic, open-source software components. The minimal IoT edge analytics stack should include a lightweight message broker, a time-series database, an ANSI-standard ad-hoc query engine, and a data visualization tool. Each component should be purpose-built for IoT.

Lightweight Message Broker

We will use Eclipse Mosquitto as our message broker. According to the project’s description, Mosquitto is an open-source message broker that implements the Message Queuing Telemetry Transport (MQTT) protocol versions 5.0, 3.1.1, and 3.1. Mosquitto is lightweight and suitable for use on all devices from low power single board computers (SBCs) to full servers.

MQTT Client Library

We will interact with Mosquitto using Eclipse Paho. According to the project, the Eclipse Paho project provides open-source, mainly client-side implementations of MQTT and MQTT-SN in a variety of programming languages. MQTT and MQTT for Sensor Networks (MQTT-SN) are lightweight publish/subscribe messaging transports for TCP/IP and connectionless protocols, such as UDP, respectively.

We will be using Paho’s Python Client. The Paho Python Client provides a client class with support for both MQTT v3.1 and v3.1.1 on Python 2.7 or 3.x. The client also provides helper functions to make publishing messages to an MQTT server straightforward.

Time-Series Database

Time-series databases are optimal for storing IoT data. According to InfluxData, makers of a leading time-series database, InfluxDB, a time-series database (TSDB), is a database optimized for time-stamped or time-series data. Time series data are simply measurements or events that are tracked, monitored, downsampled, and aggregated over time. Jiao Xian, of Alibaba Cloud, has authored an insightful post on the time-series database ecosystem, What Are Time Series Databases? A few leading Cloud providers offer purpose-built time-series databases, though they are not available at the edge. AWS offers Amazon Timestream and Alibaba Cloud offers Time Series Database.

InfluxDB is an excellent choice for a time-series database. It was my first choice, along with TimescaleDB, when developing this stack. However, InfluxDB Flux’s apparent incompatibilities with some ARM-based architectures ruled it out for inclusion in the stack for this particular post.

We will use TimescaleDB as our time-series database. TimescaleDB is the leading open-source relational database for time-series data. Described as ‘PostgreSQL for time-series,’ TimescaleDB is based on PostgreSQL, which provides full ANSI SQL, rock-solid reliability, and a massive ecosystem. TimescaleDB claims to achieve 10–100x faster queries than PostgreSQL, InfluxDB, and MongoDB, with native optimizations for time-series analytics.

TimescaleDB claims to achieve 10–100x faster queries than PostgreSQL, InfluxDB, and MongoDB, with native optimizations for time-series analytics.

TimescaleDB is designed for performing analytical queries, both through its native support for PostgreSQL’s full range of SQL functionality, as well as additional functions native to TimescaleDB. These time-series optimized functions include Median/Percentile, Cumulative Sum, Moving Average, Increase, Rate, Delta, Time Bucket, Histogram, and Gap Filling.

Ad-hoc Data Query Engine

We have the option of using psql, the terminal-based front-end to PostgreSQL, to execute ad-hoc queries against TimescaleDB. The psql front-end enables you to enter queries interactively, issue them to PostgreSQL, and see the query results.

We also have the option of using pgAdmin, specifically the biarms/pgadmin4 Docker version, to execute ad-hoc queries and perform most other database tasks. pgAdmin is the most popular open-source administration and development platform for PostgreSQL. While several popular Docker versions of pgAdmin only support Linux AMD64 architectures, the biarms/pgadmin4 Docker version supports ARM-based devices.

Data Visualization

For data visualization, we will use Grafana. Grafana allows you to query, visualize, alert on, and understand metrics no matter where they are stored. With Grafana, you can create, explore, and share dashboards, fostering a data-driven culture. Grafana allows you to define thresholds visually and get notified via Slack, PagerDuty, and more. Grafana supports dozens of data sources, including MySQL, PostgreSQL, Elasticsearch, InfluxDB, TimescaleDB, Graphite, Prometheus, Google BigQuery, GraphQL, and Oracle. Grafana is extensible through a large collection of plugins.

Edge Deployment and Management Platform

Docker introduced the current industry standard for containers in 2013. Docker containers are a standardized unit of software that allows developers to isolate apps from their environment. We will use Docker to deploy the IoT edge analytics stack, referred to herein as the GTM Stack, composed of containerized versions of Eclipse Mosquitto, TimescaleDB, and Grafana, and pgAdmin, to an ARM-based edge node. The acronym, GTM, comes from the three primary OSS projects composing the stack. The acronym also suggests Greenwich Mean Time, relating to the precise time-series nature of IoT data.

Running Docker Engine in swarm mode, we can use Docker to deploy the complete IoT edge analytics stack to the swarm, running on the edge node. The deploy command accepts a stack description in the form of a Docker Compose file, a YAML file used to configure the application’s services. With a single command, we can create and start all the services from the configuration file.

Source Code

All source code for this post is available on GitHub. Use the following command to git clone a local copy of the project.

	git clone –branch main –single-branch –depth 1 –no-tags \
	https://github.com/garystafford/iot-analytics-at-the-edge.git

view raw github_gtm.sh hosted with ❤ by GitHub

IoT Devices

For this post, I have deployed three Linux ARM-based IoT devices, each connected to a sensor array. Each sensor array contains multiple analog and digital sensors. The sensors record temperature, humidity, air quality (liquefied petroleum gas (LPG), carbon monoxide (CO), and smoke), light, and motion. For more information on the IoT device and sensor hardware involved, please see my previous post: Getting Started with IoT Analytics on AWS.

Each ARM-based IoT device is running a small Python3-based script, sensor_data_to_mosquitto.py, shown below.

	import argparse
	import json
	import logging
	import sys
	import time
	from datetime import datetime
	import paho.mqtt.client as client
	import paho.mqtt.publish as publish
	from Sensors import Sensors
	from getmac import get_mac_address
	from pytz import timezone
	# Author: Gary A. Stafford
	# Date: 10/11/2020
	# Usage: python3 sensor_data_to_mosquitto.py \
	# –host "192.168.1.12" –port 1883 \
	# –topic "sensor/output" –frequency 10
	sensors = Sensors()
	logger = logging.getLogger(__name__)
	logging.basicConfig(stream=sys.stdout, level=logging.DEBUG)
	def main():
	args = parse_args()
	publish_message_to_db(args)
	def get_readings():
	sensors.led_state(0)
	# Retrieve sensor readings
	payload_dht = sensors.get_sensor_data_dht()
	payload_gas = sensors.get_sensor_data_gas()
	payload_light = sensors.get_sensor_data_light()
	payload_motion = sensors.get_sensor_data_motion()
	message = {
	"device_id": get_mac_address(),
	"time": datetime.now(timezone("UTC")),
	"data": {
	"temperature": payload_dht["temperature"],
	"humidity": payload_dht["humidity"],
	"lpg": payload_gas["lpg"],
	"co": payload_gas["co"],
	"smoke": payload_gas["smoke"],
	"light": payload_light["light"],
	"motion": payload_motion["motion"]
	}
	}
	return message
	def date_converter(o):
	if isinstance(o, datetime):
	return o.__str__()
	def publish_message_to_db(args):
	while True:
	message = get_readings()
	message_json = json.dumps(message, default=date_converter, sort_keys=True,
	indent=None, separators=(',', ':'))
	logger.debug(message_json)
	try:
	publish.single(args.topic, payload=message_json, qos=0, retain=False,
	hostname=args.host, port=args.port, client_id="",
	keepalive=60, will=None, auth=None, tls=None,
	protocol=client.MQTTv311, transport="tcp")
	except Exception as error:
	logger.error("Exception: {}".format(error))
	finally:
	time.sleep(args.frequency)
	# Read in command-line parameters
	def parse_args():
	parser = argparse.ArgumentParser(description='Script arguments')
	parser.add_argument('–host', help='Mosquitto host', default='localhost')
	parser.add_argument('–port', help='Mosquitto port', type=int, default=1883)
	parser.add_argument('–topic', help='Mosquitto topic', default='paho/test')
	parser.add_argument('–frequency', help='Message frequency in seconds', type=int, default=5)
	return parser.parse_args()
	if __name__ == "__main__":
	main()

view raw sensor_data_to_mosquitto.py hosted with ❤ by GitHub

The IoT devices’ script implements the Eclipse Paho MQTT Python client library. An MQTT message containing simultaneous readings from each sensor is sent to a Mosquitto topic on the edge node, at a configurable frequency.

	message = {
	"device_id": get_mac_address(),
	"time": datetime.now(timezone("UTC")),
	"data": {
	"temperature": payload_dht["temperature"],
	"humidity": payload_dht["humidity"],
	"lpg": payload_gas["lpg"],
	"co": payload_gas["co"],
	"smoke": payload_gas["smoke"],
	"light": payload_light["light"],
	"motion": payload_motion["motion"]
	}
	}

view raw sensor_message.py hosted with ❤ by GitHub

IoT Edge Node

For this post, I have deployed a single Linux ARM-based edge node. The three IoT devices, containing sensor arrays, communicate with the edge node over Wi-Fi. The IoT devices could easily use an alternative communication protocol, such as BLE, LoRaWAN, or Ethernet. For more information on BLE and LoRaWAN, please see some of my previous posts: LoRa and LoRaWAN for IoT: Getting Started with LoRa and LoRaWAN Protocols for Low Power, Wide Area Networking of IoT and BLE and GATT for IoT: Getting Started with Bluetooth Low Energy (BLE) and Generic Attribute Profile (GATT) Specification for IoT.

The edge node is also running a small Python3-based script, mosquitto_to_timescaledb.py, shown below.

	import argparse
	import json
	import logging
	import sys
	from datetime import datetime
	import paho.mqtt.client as mqtt
	import psycopg2
	# Author: Gary A. Stafford
	# Date: 10/11/2020
	# Usage: python3 mosquitto_to_timescaledb.py \
	# –msqt_topic "sensor/output –msqt_host "192.168.1.12" –msqt_port 1883 \
	# –ts_host "192.168.1.12" –ts_port 5432 \
	# –ts_username postgres –ts_password postgres1234 –ts_database demo_iot
	logger = logging.getLogger(__name__)
	logging.basicConfig(stream=sys.stdout, level=logging.DEBUG)
	args = argparse.Namespace
	ts_connection: str = ""
	def main():
	global args
	args = parse_args()
	global ts_connection
	ts_connection = "postgres://{}:{}@{}:{}/{}".format(args.ts_username, args.ts_password, args.ts_host,
	args.ts_port, args.ts_database)
	logger.debug("TimescaleDB connection: {}".format(ts_connection))
	client = mqtt.Client()
	client.on_connect = on_connect
	client.on_message = on_message
	client.connect(args.msqt_host, args.msqt_port, 60)
	# Blocking call that processes network traffic, dispatches callbacks and
	# handles reconnecting.
	# Other loop*() functions are available that give a threaded interface and a
	# manual interface.
	client.loop_forever()
	# The callback for when the client receives a CONNACK response from the server.
	def on_connect(client, userdata, flags, rc):
	logger.debug("Connected with result code {}".format(str(rc)))
	# Subscribing in on_connect() means that if we lose the connection and
	# reconnect then subscriptions will be renewed.
	client.subscribe(args.msqt_topic)
	# The callback for when a PUBLISH message is received from the server.
	def on_message(client, userdata, msg):
	logger.debug("Topic: {}, Message Payload: {}".format(msg.topic, str(msg.payload)))
	publish_message_to_db(msg)
	def date_converter(o):
	if isinstance(o, datetime):
	return o.__str__()
	def publish_message_to_db(message):
	message_payload = json.loads(message.payload)
	# logger.debug("message.payload: {}".format(json.dumps(message_payload, default=date_converter)))
	sql = """INSERT INTO sensor_data(time, device_id, temperature, humidity, lpg, co, smoke, light, motion)
	VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s);"""
	data = (
	message_payload["time"], message_payload["device_id"], message_payload["data"]["temperature"],
	message_payload["data"]["humidity"], message_payload["data"]["lpg"], message_payload["data"]["co"],
	message_payload["data"]["smoke"], message_payload["data"]["light"], message_payload["data"]["motion"])
	try:
	with psycopg2.connect(ts_connection, connect_timeout=3) as conn:
	with conn.cursor() as curs:
	try:
	curs.execute(sql, data)
	except psycopg2.Error as error:
	logger.error("Exception: {}".format(error.pgerror))
	except Exception as error:
	logger.error("Exception: {}".format(error))
	except psycopg2.OperationalError as error:
	logger.error("Exception: {}".format(error.pgerror))
	finally:
	conn.close()
	# Read in command-line parameters
	def parse_args():
	parser = argparse.ArgumentParser(description='Script arguments')
	parser.add_argument('–msqt_topic', help='Mosquitto topic', default='paho/test')
	parser.add_argument('–msqt_host', help='Mosquitto host', default='localhost')
	parser.add_argument('–msqt_port', help='Mosquitto port', type=int, default=1883)
	parser.add_argument('–ts_host', help='TimescaleDB host', default='localhost')
	parser.add_argument('–ts_port', help='TimescaleDB port', type=int, default=5432)
	parser.add_argument('–ts_username', help='TimescaleDB username', default='postgres')
	parser.add_argument('–ts_password', help='TimescaleDB password', default='postgres1234')
	parser.add_argument('–ts_database', help='TimescaleDB password', default='demo_iot')
	return parser.parse_args()
	if __name__ == "__main__":
	main()

view raw mosquitto_to_timescaledb.py hosted with ❤ by GitHub

Similar to the IoT devices, the edge node’s script implements the Eclipse Paho MQTT Python client library. The script pulls MQTT messages off a Mosquitto topic(s), serializes the message payload to JSON, and writes the payload’s data to the TimescaleDB database. The edge node’s script accepts several arguments, which allow you to configure necessary Mosquitto and TimescaleDB variables.

Why not use Telegraf?

Telegraf is a plugin-driven agent that collects, processes, aggregates, and writes metrics. There is a Telegraf output plugin, the PostgreSQL and TimescaleDB Output Plugin for Telegraf, produced by TimescaleDB. The plugin could replace the need to manage and maintain the above script. However, I chose not to use it because it is not yet an official Telegraf plugin. If the plugin was included in a Telegraf release, I would certainly encourage its use.

Script Management

Both the Linux-based IoT devices and the edge node run systemd system and service manager. To ensure the Python scripts keep running in the case of a system restart, we define a systemd unit. Units are the objects that systemd knows how to manage. These are basically a standardized representation of system resources that can be managed by the suite of daemons and manipulated by the provided utilities. Each script has a systemd unit files. Below, we see the gtm_stack_mosquitto unit file, gtm_stack_mosquitto.service.

	[Unit]
	Description=GTM Stack – Mosquitto Script
	After=network.target
	[Service]
	ExecStart=/usr/bin/python3 -u sensor_data_to_mosquitto.py \
	–host ${MOSQUITTO_HOST} –port ${MOSQUITTO_PORT} –topic ${MOSQUITTO_TOPIC}
	WorkingDirectory=/home/pi/iot-analytics-at-the-edge
	StandardOutput=inherit
	StandardError=inherit
	Restart=always
	User=pi
	[Install]
	WantedBy=multi-user.target

view raw gtm_stack_mosquitto.service hosted with ❤ by GitHub

The gtm_stack_mosq_to_tmscl unit file, gtm_stack_mosq_to_tmscl.service, is nearly identical.

To install the gtm_stack_mosquitto.service systemd unit file on each IoT device, use the following commands.

	SERVICE=gtm_stack_mosquitto
	sudo cp ${SERVICE}.service /etc/systemd/system/
	sudo systemctl start ${SERVICE}.service
	sudo systemctl enable ${SERVICE}.service
	# check status
	systemctl status ${SERVICE}

view raw systemd.sh hosted with ❤ by GitHub

Installing the gtm_stack_mosq_to_tmscl.service unit file on the edge node is nearly identical.

Docker Stack

The edge node runs the GTM Docker stack, stack.yml, in a swarm. As discussed earlier, the stack contains four containers: Eclipse Mosquitto, TimescaleDB, and Grafana, along with pgAdmin. The Mosquitto, TimescaleDB, and Grafana containers have paths within the containers, bind-mounted to directories on the edge device. With bind-mounting, the container’s data will persist if the containers are removed and re-created. The containers are running on their own isolated overlay network.

	version: "3.8"
	services:
	timescaledb:
	image: timescale/timescaledb:1.7.4-pg12
	ports:
	– "5432:5432/tcp"
	networks:
	– demo-iot-net
	environment:
	POSTGRES_USERNAME: postgres
	POSTGRES_PASSWORD: postgres1234
	POSTGRES_DB: demo_iot
	deploy:
	restart_policy:
	condition: on-failure
	volumes:
	– $HOME/data/postgres:/var/lib/postgresql/data
	grafana:
	image: grafana/grafana:7.1.5
	ports:
	– "3000:3000/tcp"
	networks:
	– demo-iot-net
	deploy:
	restart_policy:
	condition: on-failure
	volumes:
	– $HOME/data/grafana:/var/lib/grafana
	user: $ID
	mosquitto:
	image: eclipse-mosquitto:1.6.12
	ports:
	– "1883:1883/tcp"
	# – "9001:9001/tcp"
	networks:
	– demo-iot-net
	deploy:
	restart_policy:
	condition: on-failure
	volumes:
	– $HOME/data/mosquitto:/mosquitto
	pgadmin:
	image: biarms/pgadmin4:4.21
	ports:
	– "5050:5050/tcp"
	networks:
	– demo-iot-net
	deploy:
	restart_policy:
	condition: on-failure
	networks:
	demo-iot-net:

view raw gtm_stack.yml hosted with ❤ by GitHub

The GTM Docker stack is installed using the following commands on the edge node. We will assume Docker and git are pre-installed on the edge node for this post.

	# on edge node
	git clone https://github.com/garystafford/iot-analytics-at-the-edge.git
	# build required directories
	mkdir -p ~/data/postgres
	mkdir -p ~/data/grafana
	mkdir -p ~/data/mosquitto/config
	mkdir -p ~/data/mosquitto/data
	mkdir -p ~/data/mosquitto/log
	# move mosquitto config
	cd iot-analytics-at-the-edge/docker/
	cp mosquitto.conf ~/data/mosquitto/config/
	# deploy stack
	docker swarm init
	docker stack deploy -c stack.yml iot
	# check status of stack
	docker stack ps iot –no-trunc
	docker stack services iot

view raw gtm_stack.sh hosted with ❤ by GitHub

First, we create the proper local directories on the edge device, which will be used to bind-mount to the container’s directories. Below, we see the bind-mounted local directories with the eventual container’s contents stored within them.

Next, we copy the custom Mosquitto configuration file, mosquitto.conf, included in the project, to the correct location on the edge device. Lastly, we initialize the Docker swarm and deploy the stack.

TimescaleDB Setup

With the GTM stack running, we need to create a single Timescale hypertable, sensor_data, in the TimescaleDB demo_iot database, to hold the incoming IoT sensor data. Hypertables, according to TimescaleDB, are designed to be easy to manage and to behave like standard PostgreSQL tables. Hypertables are comprised of many interlinked “chunk” tables. Commands made to the hypertable automatically propagate changes down to all of the chunks belonging to that hypertable.

	CREATE TABLE IF NOT EXISTS sensor_data (
	time timestamptz NOT NULL,
	device_id text NOT NULL,
	temperature double PRECISION NOT NULL,
	humidity double PRECISION NOT NULL,
	lpg double PRECISION NOT NULL,
	co double PRECISION NOT NULL,
	smoke double PRECISION NOT NULL,
	light boolean NOT NULL,
	motion boolean NOT NULL
	);
	SELECT create_hypertable('sensor_data', 'time');

view raw sensor_data.sql hosted with ❤ by GitHub

I suggest using psql to execute the required DDL statements, which will create the hypertable, as well as the proceeding views and database user permissions. All SQL statements are included in the project’s statements.sql file. One way to use psql is to install it on your local workstation, then use psql to connect to the remote edge node. I prefer to instantiate a local PostgreSQL Docker container instance running psql. I then use the local container’s psql client to connect to the edge node’s TimescaleDB database. For example, from my local machine, I run the following docker run command to connect to the edge node’s TimescaleDB database, on the edge node, located locally at 192.168.1.12.

	docker run -it –rm postgres psql \
	-U postgres -h 192.168.1.12 -p 5432 -d demo_iot

view raw docker_run.sh hosted with ❤ by GitHub

Although not always practical, could also access psql from within the TimescaleDB Docker container, running on the actual edge node, using the following docker exec command.

	TIMESCALEDB_CONTAINER=$(docker ps -q \
	–filter='name=iot_timescaledb.1' –format '{{.Names}}')
	docker exec -it ${TIMESCALEDB_CONTAINER} psql \
	-U postgres -h localhost -d demo_iot

view raw access_psql.sh hosted with ❤ by GitHub

TimescaleDB Continuous Aggregates

For this post’s demonstration, we need to create four TimescaleDB database views, which will be queried from an eventual Grafana Dashboard. The views are TimescaleDB Continuous Aggregates. According to Timescale, aggregate queries that touch large swathes of time-series data can take a long time to compute because the system needs to scan large amounts of data on every query execution. TimescaleDB continuous aggregates automatically calculate the results of a query in the background and materialize the results.

For example, in this post, we generate sensor data every five seconds from the three IoT devices. When visualizing a 24-hour period in Grafana, using continuous aggregates with an interval of one minute, we would reduce the total volume of data queried from ~51,840 rows to ~4,320 rows, a reduction of over 91%. The larger the time period or the number of IoT devices being analyzed, the more significant these savings will positively impact query performance.

A time_bucket on the time partitioning column of the hypertable is required in all continuous aggregate views. The time_bucket function, in this case, has a bucket width (interval) of 1 minute. The interval is configurable.

	— temperature and humidity
	CREATE VIEW temperature_humidity_summary_minute WITH (timescaledb.continuous)
	AS
	SELECT device_id,
	time_bucket(INTERVAL '1 minute', time) AS bucket,
	AVG(temperature) AS avg_temp,
	AVG(humidity) AS avg_humidity
	FROM sensor_data
	GROUP BY device_id,
	bucket;
	— air quality (lpg, co, smoke)
	CREATE VIEW air_quality_summary_minute WITH (timescaledb.continuous)
	AS
	SELECT device_id,
	time_bucket(INTERVAL '1 minute', time) AS bucket,
	AVG(lpg) AS avg_lpg,
	AVG(co) AS avg_co,
	AVG(smoke) AS avg_smoke
	FROM sensor_data
	GROUP BY device_id,
	bucket;
	— light
	CREATE VIEW light_summary_minute WITH (timescaledb.continuous)
	AS
	SELECT device_id,
	time_bucket(INTERVAL '1 minute', time) AS bucket,
	AVG(
	case
	when light = 't' then 1
	else 0
	end
	) AS avg_light
	FROM sensor_data
	GROUP BY device_id,
	bucket;
	— motion
	CREATE VIEW motion_summary_minute WITH (timescaledb.continuous)
	AS
	SELECT device_id,
	time_bucket(INTERVAL '1 minute', time) AS bucket,
	AVG(
	case
	when motion = 't' then 1
	else 0
	end
	) AS avg_motion
	FROM sensor_data
	GROUP BY device_id,
	bucket;

view raw view.sql hosted with ❤ by GitHub

Limiting Grafana’s Access to IoT Data

Following the Grafana recommendation for database user permissions, we create a grafanareader PostgresSQL user, and limit the user’s access to the sensor_data table and the four views we created. Grafana will use this user’s credentials to perform SELECT queries of the TimescaleDB demo_iot database.

	CREATE USER grafanareader WITH PASSWORD 'grafana1234';
	GRANT USAGE ON SCHEMA public TO grafanareader;
	GRANT SELECT ON public.sensor_data TO grafanareader;
	GRANT SELECT ON public.temperature_humidity_summary_minute TO grafanareader;
	GRANT SELECT ON public.air_quality_summary_minute TO grafanareader;
	GRANT SELECT ON public.light_summary_minute TO grafanareader;
	GRANT SELECT ON public.motion_summary_minute TO grafanareader;

view raw grafanareader.sql hosted with ❤ by GitHub

Grafana Dashboards

Using the TimescaleDB continuous aggregates we have created, we can quickly build a richly-featured dashboard in Grafana. Below we see a typical IoT Dashboard you might build to monitor the post’s IoT sensor data in near real-time. An exported version, dashboard_external_export.json, is included in the GitHub project.

Grafana’s documentation includes a comprehensive set of instructions for Using PostgreSQL in Grafana. To connect to the TimescaleDB database from Grafana, we use a PostgreSQL data source.

The data displayed in each Panel in the Grafana Dashboard is based on a SQL query. For example, the Average Temperature Panel might use a query similar to the example below. This particular query also converts Celcius to Fahrenheit. Note the use of Grafana Macros (e.g., $__time(), $__timeFilter()). Macros can be used within a query to simplify syntax and allow for dynamic parts.

SELECT
  $__time(bucket),
  device_id AS metric,
  ((avg_temp * 1.9) + 32) AS avg_temp
FROM temperature_humidity_summary_minute
WHERE
  $__timeFilter(bucket)
ORDER BY 1,2

Below, we see another example from the Average Humidity Panel. In this particular query, we might choose to validate the humidity data is within an acceptable range of 0%–100%.

SELECT
  $__time(bucket),
  device_id AS metric,
  avg_humidity
FROM temperature_humidity_summary_minute
WHERE
  $__timeFilter(bucket)
  AND avg_humidity >= 0.0
  AND avg_humidity <= 100.0
ORDER BY 1,2

Mobile Friendly

Grafana dashboards are mobile-friendly. Below we see two views of the dashboard, using the Chrome mobile browser on an Apple iPhone.

Grafana Alerts

Grafana allows Alerts to be created based on Rules you define in each Panel. If data values match the Rule’s conditions, which you pre-define, such as a temperature reading above a certain threshold for a set amount of time, an alert is sent to your choice of destinations. According to the Rule shown below, If the average temperature exceeds 75°F for a period of 5 minutes, an alert is sent.

As demonstrated below, when the temperature in the laboratory began to exceed 75°F, the alert entered a ‘Pending’ state. If the temperature exceeded 75°F for the pre-determined period of 5 minutes, the alert status changed to ‘Alerting’, and an alert was sent. When the temperature dropped back below 75°F for the pre-determined period of 5 minutes, the alert status changed from ‘Alerting’ to ‘OK’, and a subsequent notification was sent.

There are currently 18 destinations available out-of-the-box with Grafana, including Slack, email, PagerDuty, webhooks, HipChat, and Microsoft Teams. We can use Grafana Alerts to notify the proper resources, in near real-time, if an issue is detected, based on the data. Below, we see an actual series of high-temperature alerts sent by Grafana to a Slack channel, followed by subsequent notifications as the temperature returned to normal.

Ad-hoc Queries

The ability to perform ad-hoc queries on the time-series IoT data is an essential feature of the IoT edge analytics stack. We can use psql or pgAdmin to perform ad-hoc queries against the TimescaleDB database. Below are examples of typical ad-hoc queries we might perform on the IoT sensor data. These example queries demonstrate TimescaleDB’s advanced analytical capabilities for working with time-series data, including Moving Average, Delta, Time Bucket, and Histogram.

	— find max temperature (°C) and humidity (%) for last 3 hours in 15 minute time periods
	— https://docs.timescale.com/latest/using-timescaledb/reading-data#select
	SELECT time_bucket('15 minutes', time) AS fifteen_min,
	device_id,
	COUNT(*),
	MAX(temperature) AS max_temp,
	MAX(humidity) AS max_hum
	FROM sensor_data
	WHERE time > NOW() – INTERVAL '3 hours'
	AND humidity BETWEEN 0 AND 100
	GROUP BY fifteen_min, device_id
	ORDER BY fifteen_min DESC, max_temp DESC;
	— find temperature (°C) anomalies (delta > ~5°F)
	— https://docs.timescale.com/latest/using-timescaledb/reading-data#delta
	SELECT ht.time, ht.temperature, ht.delta
	FROM (
	SELECT time,
	temperature,
	abs(temperature – LAG(temperature) OVER (ORDER BY time)) AS delta
	FROM sensor_data) AS ht
	WHERE ht.delta > 2.63
	ORDER BY ht.time;
	— find three minute moving average of temperature (°F) for last day
	— (5 sec. interval * 36 rows = 3 min.)
	— https://docs.timescale.com/latest/using-timescaledb/reading-data#moving-average
	SELECT time,
	AVG((temperature * 1.9) + 32) OVER (ORDER BY time
	ROWS BETWEEN 35 PRECEDING AND CURRENT ROW)
	AS smooth_temp
	FROM sensor_data
	WHERE device_id = 'Manufacturing Plant'
	AND time > NOW() – INTERVAL '1 day'
	ORDER BY time DESC;
	— find average humidity (%) for last 12 hours in 5-minute time periods
	— https://docs.timescale.com/latest/using-timescaledb/reading-data#time-bucket
	SELECT time_bucket('5 minutes', time) AS time_period,
	AVG(humidity) AS avg_humidity
	FROM sensor_data
	WHERE device_id = 'Main Warehouse'
	AND humidity BETWEEN 0 AND 100
	AND time > NOW() – INTERVAL '12 hours'
	GROUP BY time_period
	ORDER BY time_period DESC;
	— calculate histograms of avg. temperature (°F) between 55-85°F in 5°F buckets during last 2 days
	— https://docs.timescale.com/latest/using-timescaledb/reading-data#histogram
	SELECT device_id,
	COUNT(*),
	histogram((temperature * 1.9) + 32, 55.0, 85.0, 5)
	FROM sensor_data
	WHERE temperature is not Null
	AND time > NOW() – INTERVAL '2 days'
	GROUP BY device_id;
	— find average light value for last 90 minutes in 5-minute time periods
	— https://docs.timescale.com/latest/using-timescaledb/reading-data#time-bucket
	SELECT device_id,
	time_bucket('5 minutes', time) AS five_min,
	AVG(case when light = 't' then 1 else 0 end) AS avg_light
	FROM sensor_data
	WHERE device_id = 'Manufacturing Plant'
	AND time > NOW() – INTERVAL '90 minutes'
	GROUP BY device_id, five_min
	ORDER BY five_min DESC;

view raw ad_hoc_timescale.sql hosted with ❤ by GitHub

Conclusion

In this post, we have explored the development of an IoT edge analytics stack, comprised of lightweight, purpose-built, easily deployable and manageable, platform- and programming language-agnostic, open-source software components. These components included Docker containerized versions of Eclipse Mosquitto, TimescaleDB, Grafana, and pgAdmin, referred to as the GTM Stack. Using the GTM stack, we collected, analyzed, and visualized IoT data, without first shipping the data to Cloud or other external systems.

---

This blog represents my own viewpoints and not of my employer, Amazon Web Services. All product names, logos, and brands are the property of their respective owners.