GCP IIoT: Streaming IoT Data with Google Cloud and FogLAMP

An end-to-end demo to build a data management platform for IoT data with Google Cloud and FogLAMP

Photo by Anna Nekrashevich from Pexels

Introduction

We are going to illustrate how to build an Industrial IoT data management platform on Google Cloud (GCP). In this post we will focus on first mile data collection using FogLAMP, storage, real time analytics and alerting using Google Dataflow and BigQuery. In the next post we will show how to visualize the device inventory and analytics using Looker dashboards, and finally how to leverage the power and simplicity of BigQuery ML to perform anomaly detection — stay tuned.

Before we begin, IoT can be a loaded term, so let’s break it down a bit. From a technical perspective, IoT is just a specific use case of a more generic time-series collection and analytics problem — it has a lot in common with market data analytics, system metrics monitoring and alerting, etc. (There are also a lot of differences, we will discuss those later)

The problem can be broken down into 4 stages:

1. First-mile data collection — Data needs to be collected in a secure way from devices in the field and stored somewhere as a time-series of measurements. We are using FogLAMP for that.
2. Analytics & ML — The measured time-series data needs to be analyzed to derive business insights and actions. GCP has a proven set of tools and services that have been proven to solve time-series problems at Google scale.
3. Storage — The time-series data must be stored and archived for audibility and historical analysis. Google BigQuery provides cheap, scalable and fault tolerant storage.
4. Dashboarding — Device topology and metadata, measurement time-series data and the derived analytics need to be displayed in dashboards to business users and stakeholders. We will show how to use Looker to create near real-time self-serve dashboards in the next post.

The architecture and code provided in this post form the foundation for a turn-key, next generation data historian that leverages GCP’s scale, compute power, and advanced ML and Analytics to provide a much cheaper, flexible and powerful solution than existing products.

What is IoT?

The term “Internet of Things” (IoT) is used to refer to any network of machines and devices that are connected to physical objects and can collect information from their environment, such as temperature, pressure, etc., as well as automate and optimize the execution of processes. The manufacturing and utilities sectors leverage the advantages of IoT in applications such as quality control, inventory management, predictive maintenance, and smart metering, among others¹. While the Industrial IoT (IIoT) spending has expanded in recent years², there are a variety of challenges to overcome³.

In large industrial operations with multiple locations, IoT devices can be trapped inside “sensor silos” with limited connection to a centralized data management system. Collecting the data from such sources can be a tedious manual process. Moreover, IoT networks often involve devices with heterogeneous Operational Technology (OT) protocols that make it harder to integrate them together. These are just a few examples of remaining obstacles that make it challenging to integrate and analyze IIoT data from multiple systems.

However, when data from IoT devices are collected efficiently and securely, they can help industrial companies achieve digital transformation by delivering real-time information that stakeholders use to quickly gain insights and make critical decisions. For instance, operations managers can leverage that data to detect process anomalies in real-time, whilst accounting teams can gain an accurate picture of the balance between the inputs and sales of a production facility.

In this article, we will look at how FogLAMP can be used to centralize the management of IIoT devices and data. We will also walk you through the approach in our demo for leveraging FogLAMP with Google Cloud Platform (GCP) and Apache Beam to ingest, process, and store IIoT data at scale.

What is FogLAMP?

One of the biggest challenges that industrial IoT users face is collecting heterogeneous data from a plethora of different sensor protocols and industrial systems, and delivering them to stakeholders that can turn these data into useful insights. The bulk of measurement data often never leaves the premises of an industrial facility, except by manual processing, which hinders the ability of businesses to make timely decisions from operations data. This is also known as the “First Mile Problem”. FogLAMP is an open-source framework that was designed by Dianomic to solve that problem. It provides a scalable and secure infrastructure for centralizing the collection, buffering, and routing of IIoT sensor data to historian and other management systems on premise or on the cloud. FogLAMP instances are implemented on the edge as a collection of services that enable core functionalities (security, monitoring, and storage), as well as data processing (transformation, alerting, collection, and transmission). Although FogLAMP can be used “as-is” to connect to a wide range of sensor hardware types and sensor protocols, its open-source license and modular architecture mean that developers can expand it by writing new Python plugins to connect to devices and data management systems that are not supported by default. Dianomic has recently announced FogLAMP Manager, a platform for users to securely provision and manage all of their edge IIoT connections, devices, and machine learning applications. Centralized management is the primary feature of FogLAMP Manager, while also offering a range of other features that enhance the experience of IIoT users, such as templating scoped for different meter types, configuration management and version control, asset hierarchy management, role-based access control, and audit logging.

Image credit: FogLAMP

Why FogLAMP and GCP?

Collecting the data is only the first step — to make it truly valuable we need to place it in a single location that is accessible from anywhere. Traditionally, the storage and analysis of IoT data has been performed in specialized timeseries databases designed to store process data, often referred to as “Data Historians’’. However, historians were originally designed to operate at single plant scale, and while some vendors offer solutions that can scale across plants, they cannot rival the scale of a true cloud solution and force customers into an expensive vendor lock.

The combination of FogLAMP’s unique proposition for orchestrating industrial edge applications and Google’s cloud services offer a complete stack to deliver new experiences with cloud capabilities at the edge. The two integrate seamlessly thanks to the Internet of Things (IoT) Core, Google’s fully managed service for securely connecting IoT devices. The ingested data can then be consumed in any of the available services on Google Cloud:

1. Dataflow/Dataproc: efficient parallel processing for massive amounts of streaming and batch data
2. BigQuery and BigQuery ML: serverless, highly scalable data warehouse for storage, analytics, and machine learning models
3. AutoML: an out-of-the-box AI interface for building custom machine learning models. This is an excellent complement to FogLAMP’s edge AI applications based on TensorFlow Lite that enable engineers and operations managers to quickly spin out machine learning models on their edge devices.
4. Vertex AI: machine learning operations (MLOps) tool for managing data and models
5. Looker: dashboards and analytics

Users can also take advantage of the combined power of Dataflow, BigQuery, and Datastream to enrich their process data by importing data from literally any external source whether it’s company-owned or open-source, unlike existing historians that are usually limited to only a few database options. Finally, businesses can also leverage the products of Independent Software Vendors (ISVs) and open-source frameworks that offer their specialized edge applications on Google Cloud.

Event and Alerting Framework

Businesses are often interested in monitoring activities that are significant to a process or operation. When these activities have a clear start and end time are often referred to as events or event frames. Examples of events include equipment downtime, transient environmental changes, IoT device outages, etc. The event data need to be captured and be readily combinable with regular process data to enable analytics and alerting. Moreover, engineers and operations managers need a platform that allows them to define the events that are important to their processes, automatically capture the real-time event data, and associate sequences of different events to identify root causes.

In our solution, there are some common pre-configured events, such as capturing version changes of IoT devices and device connection losses. Additional user-defined event criteria are stored in BigQuery. The users can specify the type of event, the devices or equipment that it applies on, as well as define the event trigger conditions by setting a baseline value and an operator. For example, a gas/oil pipeline operator might be interested in capturing events caused by critically low flowrate or high temperature:

Event definitions table.

The table with the event criteria is imported to the Dataflow workers that combine it with the real-time stream of raw sensor data to automatically identify the data points that are part of an event. The event data and the process data are stored in different tables but in the same BigQuery dataset so that they can be easily combined together for downstream analysis:

Event data table.

Using SQL queries in BigQuery, the users can create summary tables of events with critical attributes about each event, such as severity and start/end times:

Event summary table.

How the Cloud Architecture Works

Cloud Architecture overview.

There are four main framework units in the proposed architecture:

1. We operate a FogLAMP instance on a virtual machine hosted on GCP to transmit real-time IIoT sensor data. FogLAMP can run on any Linux derivative system, in containers, or on bare metal systems, but for this demo we opted to host it on the cloud. IIoT sensors are simulated using the Prosys OPC UA Simulation Server hosted in the same VM as the FogLAMP instance.
2. For ingesting data from the FogLAMP instance, we use the Google Cloud Internet of Things (IoT) Core. The FogLAMP GCP plugin connects to the IoT Core using MQTT and is compliant with the security requirements of GCP. A Pub/Sub topic receives the streaming data from the IoT Core and serves as the point-of-entry for the integration with the rest of the GCP services.
3. For parallel data processing, we run Apache Beam on Google Cloud Dataflow workers. A combination of different Apache Beam Pipelines are designed to parse the raw IIoT data, capture critical event data, and compute timeseries metrics, such as moving average and relative strength index. To that end, we have leveraged in our pipelines Beam’s State and Timer APIs, as well as the Dataflow Timeseries Library.
4. Finally, the processed data from Dataflow are stored in tables hosted on BigQuery, Google’s serverless and scalable near-real time data warehouse. Each measurement entry is appended to that table with columns that include timestamp, device ID, and property measured, among others:

Raw measurements table.

Similarly, the metrics computed using the Dataflow Timeseries Library are gathered in a separate table:

Timeseries aggregated metrics table.

Capturing Process Events with the Apache Beam State and Timers

A typical IIoT measurement data point has at least two keys: a major key that is usually the sensor unique identifier and a minor key that is usually the property measured. In our Dataflow pipelines, we used the Timeseries Library to compute aggregated metrics of the properties being measured, as well as Beam’s State and Timer APIs to analyze the inbound stream of IoT data for events.

The raw sensor data are parsed in a Dataflow pipeline where they are combined with a side-input table from BigQuery that contains the conditions for each event type:

The event measurements are then forwarded to a separate Dataflow pipeline that is dedicated for processing events. The challenge that we faced was how to identify the measurements that were part of the same event, while also streaming these measurements to BigQuery in real-time. The first thought that came to mind was to use session windowing in Beam. The problem with that approach is that an element is not assigned to a session when it is first processed. Instead, each element is placed into a single proto-session window that begins at that record’s event time and spans the width of the session gap duration. Then, any windows that are encountered in the future that overlap this window will be grouped together and emitted once the window closes, which could be hours for a long-lasting event, such as equipment downtime. To overcome that, we decided to leverage the State and Timer APIs. We define a @ProcessElement block, with state and timers that are each scoped to a key consisting of the sensor ID and the property measured. The first time an element (i.e. an event measurement) enters the @ProcessElement block it will be assigned a random UUID scoped to that key and write that value to the event ID state. Every subsequent element will reset the timer and the block will emit that element after appending to its value the current event UUID from the state. After a user-defined duration without any new event measurements for a given key, the timer for that key will expire and the @OnTimer callback will be triggered and write a new UUID to the event ID state. That @ProcessElement block can be implemented as follows:

Another common requirement in industrial IoT use cases is the ability to detect device outages and have stopped transmitting data. In our main Dataflow pipeline, we define a @ProcessElement block that accepts state values and a timer, which are each scoped to the key of the timer being handled. The first element that enters the block will write the value of the key (in this case the sensor ID) to the key StateValue. Every subsequent element will reset the timer and we output the value of that element. After a user-defined duration without any new elements for a given key, the timer for that key will expire and the @OnTimer callback will be triggered and output a value that indicates that this device has gone silent. That value will now be streamed to the BigQuery sink and can be used to activate notifications and visualize on dashboards. The @OnTimer callback will keep looping, emitting these values until a new element is received for that key. Here is an example of how to implement the State and Timer in Beam to emit mock rows when a device goes silent:

The use of the State and Timer API in our pipelines is best illustrated in the following diagram:

Demo Implementation

Here you can find our demo project on GitHub. The detailed, step-by-step instructions on GitHub will take you through running Terraform on your local machine, which will provision a virtual machine with FogLAMP, the IoT Core and Pub/Sub topics used for data ingestion, a BigQuery dataset with sample tables, the Dataflow jobs, and the RSA keys required for authenticating your local machine and FogLAMP to GCP. The instructions will also walk through on how to access FogLAMP’s GUI using Chrome Remote Desktop to set up your sample telemetry device. As soon as you activate your virtual device, you will be able to explore the live data streaming into the BigQuery tables.

Conclusion

For many businesses that rely on IIoT to gain operational insights, centralizing the management, analysis, and integration of heterogeneous measurement devices and data can be a significant challenge to overcome. This article showcased a scalable architecture based on Google Cloud, FogLAMP, and Apache Beam for managing IIoT devices and ingesting measurement data to a centralized database for storage, as well as capturing user-defined event frames from the ingested IIoT data. By centralizing their IIoT processes with FogLAMP and Google Cloud, companies will realize benefits over existing historians in several areas:

1. Robust decision-making by leveraging a wide variety of real-time operational data that can be readily combined with data from external sources
2. Flexibility without vendor locks to design a software stack that truly meets a company’s needs rather than applying an one-size-fits-all solution
3. Scalable computing power, storage, and cost that adapt to the company’s growth
4. Improved data auditability and quality
5. Create a strong foundation to support future growth by deploying technologies that are scalable across a company’s assets and by readily applying emerging technologies such as ML/AI to drive insights and optimize process operations.