This post is a revised version of an earlier post, featuring major version updates of TimescaleDB (v1.7.4-pg12 to v2.0.0-pg12), Grafana (v7.1.5 to v7.5.2), and Mosquitto (v1.6.12 to v2.0.9). All source code and SQL scripts are revised. Note that TimeScaleDB has a current limitation/bug with Docker on ARM later than v2.0.0.

The Edge

Edge computing is a fast-growing technology trend, which involves pushing compute capabilities to a network’s 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 close to the data source 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.

David Ricketts, Head of Marketing at Quiss Technology PLC, in his post, Cloud and Edge Computing — The Stats You Need to Know for 2018, estimates that 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 their 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 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 a 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.

Edge devices such as IoT Gateways, which aggregate and transmit IoT data from these devices to external systems, are generally lower-powered, with limited processors, memory, and storage. 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 machine learning (ML) inference capabilities to edge devices. Azure offers Azure IoT Edge. Azure IoT Edge provides the ability to run artificial intelligence (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 real-time, streaming IoT data at the edge is critical to a rapid feedback loop. IoT edge analytics can accelerate anomaly detection and remediation, improve predictive maintenance capabilities, and expedite proactive inventory replenishment.

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:

1. Lightweight message broker;
2. Purpose-built time-series database;
3. ANSI-standard SQL ad-hoc query engine;
4. Data visualization tool;
5. Simple deployment and management framework;

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-powered 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 light-weight 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 architecture 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 is designed for performing analytical queries, both through its native support for PostgreSQL’s full range of SQL functionality and 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 Grafana, TimescaleDB, Eclipse Mosquitto, and pgAdmin, to an ARMv7-based edge node. The acronym, GTM, comes from the three primary OSS projects composing the stack. The abbreviation 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. Note that the updated version of the source code for this post is in the v2021–03 branch.

git clone --branch v2021-03 --single-branch --depth 1 \
    https://github.com/garystafford/iot-analytics-at-the-edge.git

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
Analyze environmental sensor data from IoT devices in near real-time with AWS IoT Analyticstowardsdatascience.com

Each ARM-based IoT device runs 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.publish as publish
	from getmac import get_mac_address
	from pytz import timezone
	from Sensors import Sensors
	# Sensor to Mosquitto Script
	# Author: Gary A. Stafford
	# Date: 2021-03-26
	# 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, hostname=args.host, port=args.port)
	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

Below are the actual sensor readings sent by the IoT device as an MQTT message to the Mosquitto topic.

	{
	"data": {
	"co": 0.0031827073092533685,
	"humidity": 51.099998474121094,
	"light": true,
	"lpg": 0.005553622262501496,
	"motion": false,
	"smoke": 0.01449612738171321,
	"temperature": 19.100000381469727
	},
	"device_id": "00:0f:00:70:91:0a",
	"time": "2021-04-02 17:23:44.809046+00:00"
	}

view raw sample_message.json 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. 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 also runs 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
	# Mosquitto to TimescaleDB Script
	# Author: Gary A. Stafford
	# Date: 2021-03-31
	# 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

Like 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 the necessary Mosquitto and TimescaleDB connection settings.

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 can 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 Linux-based IoT devices and edge nodes 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 objects that systemd knows how to manage. This is 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 file. Below, we see the gtm_stack_mosquitto unit file, gtm_stack_mosquitto.service.

	[Unit]
	Description=GTM Stack – Sensor to Mosquitto Script
	After=network.target
	[Service]
	ExecStart=/usr/bin/python3 -u sensor_data_to_mosquitto.py \
	–host "192.168.1.12" –port 1883 –topic "sensor/output"
	WorkingDirectory=/home/pi/iot-analytics-at-the-edge/scripts
	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 systemctl/${SERVICE}.service /etc/systemd/system/
	sudo systemctl start ${SERVICE}.service
	sudo systemctl enable ${SERVICE}.service
	# check status
	systemctl status ${SERVICE}.service
	ps aux | grep sensor_data_to_mosquitto.py

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, Grafana, and 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 configuration and data will persist if the containers are removed and re-created. The containers are running on an isolated overlay network.

	version: "3.9" # optional since v1.27.0
	services:
	timescaledb:
	image: timescale/timescaledb:2.0.0-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.5.2
	ports:
	– "3000:3000/tcp"
	networks:
	– demo-iot-net
	deploy:
	restart_policy:
	condition: on-failure
	volumes:
	– "$HOME/data/grafana:/var/lib/grafana"
	user: $ID:1
	mosquitto:
	image: eclipse-mosquitto:2.0.9
	ports:
	– "1883:1883/tcp"
	networks:
	– demo-iot-net
	deploy:
	restart_policy:
	condition: on-failure
	volumes:
	– "$HOME/data/mosquitto/config:/mosquitto/config"
	– "$HOME/data/mosquitto/data:/mosquitto/data"
	– "$HOME/data/mosquitto/log:/mosquitto/log"
	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 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 will create several local directories on the edge device, which will be used to bind-mount to the Docker 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 edge device’s correct location.

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 and 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 as practical, you can 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 will create four TimescaleDB materialized views, which will be queried from a Grafana Dashboard. The materialized views are TimescaleDB Continuous Aggregates. According to Timescale, aggregate queries which 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. To make these queries faster, a continuous aggregate allows materializing the computed aggregates, while also providing means to continuously, and with low overhead, keep them up-to-date as the underlying source data changes.

For example, we generate sensor data every five seconds from the three IoT devices in this post. 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 for all continuous aggregate views. The time_bucket function, in this case, has a bucket width (interval) of 1 minute. The interval is configurable.

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

view raw materialized_views.sql hosted with ❤ by GitHub

To automatically refresh the four materialized views, we will create four corresponding continuous aggregate policies. In this demonstration, the continuous aggregate policies create a refresh window between one week ago and one hour ago, with a refresh interval of one hour.

	— create policies that automatically refreshes continuous aggregates
	SELECT add_continuous_aggregate_policy('air_quality_summary_minute',
	start_offset => INTERVAL '1 week',
	end_offset => INTERVAL '1 hour',
	schedule_interval => INTERVAL '1 hour');
	SELECT add_continuous_aggregate_policy('light_summary_minute',
	start_offset => INTERVAL '1 week',
	end_offset => INTERVAL '1 hour',
	schedule_interval => INTERVAL '1 hour');
	SELECT add_continuous_aggregate_policy('motion_summary_minute',
	start_offset => INTERVAL '1 week',
	end_offset => INTERVAL '1 hour',
	schedule_interval => INTERVAL '1 hour');
	SELECT add_continuous_aggregate_policy('temperature_humidity_summary_minute',
	start_offset => INTERVAL '1 week',
	end_offset => INTERVAL '1 hour',
	schedule_interval => INTERVAL '1 hour');
	—view jobs
	SELECT * FROM timescaledb_information.jobs;
	—view job stats
	SELECT job_id, total_runs, total_failures, total_successes
	FROM timescaledb_information.job_stats;

view raw add_continuous_aggregate_policy.sql hosted with ❤ by GitHub

Advanced Analytic Queries

The ability to perform ad-hoc queries on time-series IoT data is an essential feature of the IoT edge analytics stack. We can use psql, pgAdmin, or even our own IDE to perform ad-hoc queries against the TimescaleDB database on the edge node. Below are examples of typical ad-hoc queries a data analyst might perform on 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.

	— ad-hoc queries
	— 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(time),
	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
	WITH ht AS (SELECT time,
	temperature,
	abs(temperature – lag(temperature) over (ORDER BY time)) AS delta
	FROM sensor_data)
	SELECT ht.time, ht.temperature, ht.delta
	FROM 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(time),
	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 adhoc_timescaledb.sql hosted with ❤ by GitHub

Data Visualization with Grafana

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.

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

Using PostgreSQL in Grafana

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

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 Celsius 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 limit the humidity data to a valid 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 the 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 laboratory temperature 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 changes to ‘Alerting’, and an alert is 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 twenty alert notifiers available out-of-the-box with Grafana, including Slack, email, PagerDuty, webhooks, VictorOps, Opsgenie, 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 the Slack channel, followed by subsequent notifications as the temperature returned to normal.

Conclusion

This post 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 Grafana, TimescaleDB, Eclipse Mosquitto, 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 (AWS). All product names, logos, and brands are the property of their respective owners.