Analysis and prediction of big stream data in real-time water quality monitoring system

Abstract

Large scale real-time water quality monitoring system usually produces vast amounts of high frequency data, and it is difficult for traditional water quality monitoring system to process such large and high frequency data generated by wireless sensor network. A real-time processing and early warning system framework is proposed to solve this problem, Apache Storm is used as the big data processing platform, and Kafka message queue is applied to classify the sample data into several data streams so as to reserve the time series data property of a sensor. In storm platform, Daubechies Wavelet is used to decompose the data series to obtain the trend of the series, then Long Short Term Memory Network (LSTM) model is used to model and predict the trend of the data. This paper provides a detailed description concerning the distribution mechanism of aggregated data in Storm, data storage format in HBase, the process of wavelet decomposition, model training and the application of mode for prediction. The application results in Xin’an River in Yantai City reveal that the prosed system framework has a very good ability to model big data with high prediction accuracy and robust processing capability.

Keywords

Water quality monitor system wavelet analysis deep learning big data process prediction of time series LSTM

1. Introduction

Sustainable development of socio-economics and environment requires protecting water quality and reasonably allocating water resources among different users. Wireless Sensor Network (WSN) is widely applied in the field of hydrology and water conservancy [20]. Different from traditional manually collecting data samples for water quality monitoring and analysis, WSN for water quality monitoring is an online method with the following advantage characteristics [1]: (1) intensive sampling; (2) real-time and in-time; (3) distributed acquisition; and (4) longer life span. Sensors are deployed in the monitored waters to collect parameters such as pH, Dissolved Oxygen (DO), water temperature, flow rate and so on. Through converging gateway, data of these parameters are uploaded to upper computers, where they are processed and analyzed.

The whole process of WSN of water quality monitoring is usually divided into four phases: (1) Data sampling; (2) Data transmission; (3) Data processing and analyzing; and (4) Result display. The functions of water quality monitoring systems may be varied due to different business needs and system hardwares. Some systems are required to collect a single index while others may be used to collect multiple indices. The sampling periods may also vary from seconds, minutes, hours to days, or even longer. The obtained data are more accurate when the sampling frequency is higher, but meanwhile the sensor nodes will require higher energy consumption [21]. Therefore, most existing systems have a sampling frequency of 24 times/day [28] and even lower ones. Low frequency can reduce node power consumption and prolong the life cycle of the network, while it is not accurate enough for drinking water detection and sudden pollution alert.

In contrast, if high sampling frequency is adopted, massive data will be generated in large-scale water quality monitoring network. The data generated by a sensor is a time series, which is a grim challenge for a traditional relational database and stand-alone operation data processing system, both in data storage and in real-time data analysis [12]. Consequently, the usage of water quality monitoring systems is limited due to its insufficient abilities of sampling large data series and historical data inquiry. Monitoring systems usually make early warning through analysis of the data changes between current and previous time. However, changes of the sampling data can be caused by the errors of the sensor itself and the instantaneous interference of water quality. Such variations cannot accurately reflect the changes of water quality. In addition, the existing early warning mechanism is realized by regression models and other inaccurate functions, which also result in large errors.

Therefore, the use of a typical WSN for large-scale water quality monitoring has deficiencies in the following four aspects:

The sampling frequency is relatively low;

Data processing efficiency is difficult to meet the real-time requirements;

The error of the regression model is too large to accurately simulate the reality;

Data prediction and early warning functions are not perfect.

A real-time system to monitor and predict the water quality was developed, aiming at solving those four issues. In our system, IoT (Internet of Things) devices are used to collect the data, and Apache storm is adopted as the data process platform to deal with the data stream generated by the sensor network.

2. Related works

Many solutions to water quality monitoring using IoT technology have been developed since the early 21st century. Zennaro et al. implemented an application of IOT to measure water quality in Malawi [28]. In the system, the sensor nodes took water quality samples at a pre-defined day time interval, and then the sampled data was sent and stored in the gateway station. The sensor nodes would go to sleep after sending the data, and they waked up after a fixed period. In this application, the frequency of data sampling was relatively low, and MySql was used as storage engine for the purpose of data acquisition and statistical analysis. Krish et al. proposed a Supervisory Control and Data Acquisition (SCADA) system integrated with the IoT technology for real-time water quality monitoring [13], which was applied in the Tirunelveli Corporation (Metro city of Tamilnadu state, India). SCADA automatically measured the parameters of pressure, pH, level, and energy by sensors using Arduino Atmega 368, and the sensor data was automatically sent through Global System for Mobile Communication (GSM). SCADA stored the data in DB(Data Base) for data acquisition in the system. Vijayakumar et al. proposed a real time water quality monitoring system [23], in which several sensors were used to measure the physical and chemical parameters of the water, such as temperature, pH, Turbidity, Conductivity and DO. The measured values from the sensors were processed by the raspberry PIB+ model based core controller, and then they could be viewed on internet using cloud computing. Madhavireddy et al. proposed a Smart Water Quality Monitoring System Using IoT technology [16]. The system used microcontroller and the processed values that was remotely to the core controller ARM through a WI-FI protocol to interface the measured values from the sensors [16]. The data continuously updated at certain intervals in the server could be retrieved or accessed from other places in the world. Furthermore, it provided an alert mechanism, where a buzzer would be ON if the sensors do not work or get into abnormal conditions.

However, these systems only focus on real time water quality monitoring, and unfortunately they lack the functions of forecasting and early warning the future situations of water quality. Whereas predicting the situations of future water quality plays an even more vital role for planning and pollution control. However, water quality forecasting needs to deploy sensor nodes widely so as to obtain the parameters of a whole area rather than the local area. Deploying nodes widely will generate great amount of continuous data at high speed, named big data stream. The traditional software platforms cannot deal with such high speed big data stream.

To process these vast amounts of high frequency data, MapReduce has been widely used in cluster environment, where it provides data parallel processing ability to meet the throughput and real-time requirements [11]. Apache Storm is a distributed real-time big data-processing system, and it is designed to process large amount of data in a fault-tolerant and horizontal scalable method. It is a streaming data framework that has the capability of highest ingestion rates [3]. In IoT based monitoring system, the multisource heterogeneous sensor data integration and streaming sensor data processing in real time manner with low latency are key issues [10]. Zhou et al. proposed an efficient streaming spatio-temporal data access based on Apache Storm (ESDAS) to achieve real-time streaming data access and data cleaning [29]. The consumption time in ESDAS is approximately 300 milliseconds for ten thousand records per second for a speeding bolt, and its efficiency is approximately three times higher than that of MongoDB. A Click- Stream Process System (CSPS) was proposed by Jonathan, which focused on the analysis of Wikipedia Clickstream data with Apache Storm. CSPS showed that increasing number of parallel thread would improve throughput. Particularly, it is possible to efficiently stream and analyze stream data, which allows even greater flexibility in prioritizing trade-offs and performance properties of a system [19].

In order to make early warning of the water quality problems, it need get the trend of water quality parameter rather than using the original sample data. Wavelet analysis is a well-known tool used to get the trend of a time series, which decomposes a signal into approximations and details. The approximations are the high-scale, low- frequency components of the signal, from which we can get the trend of the constant data [14]. Consequently, the trend can be used to predict the water parameters with deep learning technology [27]. Wei et al. proposed an integrated approach by using discrete wavelet multi-resolution decomposition and a back- propagation (BP) feed-forward multilayer perceptron (FFML) artificial neural network (ANN) [25]. In their experiment, monthly flow data from three gauges in the Weihe River in China were used for network training and testing for 48-month-ahead prediction, and the result showed that the model was able to significantly increase the prediction accuracy. However, only small amount of data was used in the experiment, thus it need further prove if this method is able to process real time big data stream. Guillen-Navarro et al. developed an intelligent framework based on a deep learning model for early prediction of crop frost to help farmers activate anti-frost techniques in order to save the crop [7]. This framework used LSTM based model to predict low temperature using information from an IoT infrastructure deployed on two plots in Murcia (Southeast of Spain). Three experiments results showed that the average quadratic errors less than a Celsius degree and the determination coefficient $R^{2}$ is greater than 0.95. Similarly, whether it was able to deal with big data was not discussed in the work.

This research complements the current studies in the following two main aspects: (1) A big data platform was developed to process the real time data stream from wireless sensors at temporal granularity of seconds; (2) A wavelet-LSTM model was developed to predict the future trend of water quality and make an early warning for the future water quality problems.

3. Sytem design

Xin’an River in Yantai, China was chosen as the experimental environment. For the purpose of the experiment, only one monitoring point in a section of the river was deployed. According to the business needs of water quality monitoring, the entire system consists of four parts: the wireless sensor network, the data receiving server, the Apache Storm data processing system and the Web-based data display system. The overall system framework is shown in Fig. 1.

Fig. 1.

System framework.

IoT computing concept describes the connection of physical objects to the Internet with intelligence to identify themselves from other devices, and it can get the information of physical world and send it to a data process platform. According to the typical IoT architecture, our system is composed of three layers, namely perception layer, network layer and application layer from bottom to top. The perception layer comprises of sensors, the network layer contains standard communication protocol to process information on the required network and application layer. The most important feature of IoT are sensors, which are used for capturing environment information through the connection with online network using specific protocols for communication and information exchange. Currently, the system deploys three types of sensors for getting the parameters of pH, DO and water temperature in the monitored water. The measurement of pH in a water system is of utmost importance, because it is used to diagnose whether the water is acidic or alkaline. Dissolved Oxygen (DO) is another important parameter of water quality, which measures the concentration of non-compound oxygen in water. Too large and too small DO level will degrade the quality of water and cause damage to aquatic life. Besides the chemical parameters, water temperature is another critical factor that affects DO levels, the solubility and reaction rates of chemicals, metabolism, growth, and reproduction, Species composition, Water density and stratification, and whatever. In this connection, water temperature sensor was also adopted in our system.

The perception layer is consisted of WSN nodes, which are generally powered by batteries, and IEEE 802.15.4 is used as the MAC layer to save energy [13]. The typical WSN is used for low-frequency event monitoring, which does not require high real-time data, so it can meet the needs of general industrial monitoring [24]. In the water quality monitoring system, however, water pollution in many cases is an emergency event, which causes spread in a large area in a short time. In this connection, the monitoring system with a sufficiently high sampling frequency is required. Whereas battery power supply cannot satisfy with such requirement. In this case, wireless transmission terminals are used as communication gateways, and solar energy, wind energy, and storage batteries are used as energy supplies in our system. Besides, sensors and wireless gateways are connected by the 485 Bus, and the sensors send the parameters they acquired to the gateway periodically through the Bus. After receiving the parameters, the gateway transmits them to our data center immediately. The monitoring site is too far away from the data center, hence the wireless gateway uses a 4G mobile communication network. In the data center, a socket server response for data receiving and an Apache storm platform is for data analysis and prediction.

In the monitoring system, the data of each monitoring point should be processed to retain the data time attribute, i.e. a continuous data stream, thus it is especially suitable to use the Apache Storm cluster for the process. In order to obtain higher monitoring and prediction accuracy, Trident is used in the system to ensure the data transaction process.

The system consists of two software components, one is a web application for basic data setup, data display and query, and the other is a data processing system. In the data processing system, the data is sent to the Storm cluster through the Socket server, and the Kafka message queue is used as the data communication channel in the middle. To guarantee the real-time and transactional nature of data stream processing, Trident is used in Storm to receive data from Kafka, and then wavelet analysis is used to obtain the trend of the data. LSTM based model is employed to predict the future water quality based on the wavelet decomposed trend. Finally, the data processing results are send to the Web system presentation page through WebSocket service. The software system framework is displayed in Fig. 2.

Fig. 2.

Software architecture.

4. Key problems

In the course of system design, we focused on solving the following problems:

4.1. Serial data parallelization

Apache Storm is a distributed real-time big data processing system, and it is designed to process vast amount of data in a fault-tolerant and horizontal scalable method. It is a streaming data framework with the capability of highest ingestion rates. By utilizing Storm, it is convenient to execute all kinds of manipulations on real-time data in parallel.

In the view of server, the data from a gateway is data stream in serialized form, but it is composed by the data from several types of sensors. The data stream will be decomposed by wavelet to process the parameters respectively, thus Trident is used to accomplish this process. Trident, an extension of storm, is a high-level abstraction for doing real-time computing on top of Storm, and it can seamlessly intermix high throughput (millions of messages per second), stateful stream processing with low latency distributed querying.

Stream is the core data model of Apache Storm trident, and it is handled as a series of batches. Apache Storm can receive data from many spouts as the source. Firstly, spouts are setup according to the external data source, such as Kafka. Then the streams are created with the same number of spouts. Finally, the streams are merged to one. By using stream, data is treated as a series of batches. Between the nodes in the Storm cluster, Trident handles the input stream and converts the input continuation into consecutive batches containing several tuples to process. Through redirection, each batch is distributed to different partitions for parallelization of data processing [4,22]. If random redirection is used, it will inevitably lead the data from the same sensor being distributed to different partitions, and then to be aggregated, thus the time series attribute of the sensor data will be destroyed and the trend judgment result and the accuracy of the prediction will be affected.

In the system, we create a Topic for each monitoring area in Kafka. The Kafka producers in the Socket Server partition the data according to the monitoring point numbers, and each data carries the sensor number of the monitoring point. Storm creates a consumer Spout with the same numbers of partitions, receives Kafka data and generates a data stream, and then distributes the data stream based on the sensor numbers. In this way, the advantages of Kafka and Storm are collectively used to parallelize the sensor data, and the time sequence of the data is maintained by fully utilizing the cluster computation. The data decomposition process is shown in Fig. 3. For each wireless gateway of the monitoring areas, the system has configured the area number and the monitoring point number. When the wireless gateway forwards sampling data, the water area number, monitoring point number, sensor number, sampling time, and sampling value constitute a data packet.

Fig. 3.

Procedure of data decomposition.

The producer of Kafka sends data to the corresponding TOPIC according to the water area number, and divides data into partitions according to the monitoring point number. The format of the data sent by the gateway is shown as following: $\begin{array}{l} areaId + stationId + sensorId + timestamp \\ + sensorValue \end{array}$ At the data center, socket server receives the data and parses it. The basic information table is queried with the keyword ‘sensorld’ to get the type of the sensor, then the value of the sensor is stored into corresponding table.

In Storm, the number of Worker of the Topology is established according to the number of TOPIC, while the parallelism of Spout is created based on the number of TOPIC partitions. The data is distributed by the partitionBy method of Storm stream in accordance with the sensor number, and the sampling data of each sensor is processed with a separate thread.

4.2. Wavelet decomposition

In the water quality monitoring system, the data sampled from an index sensor are usually used to compare with a pre-defined threshold to determine whether the value is normal [1]. Due to the influence of external factors in natural environment and temporary changes of the monitored indices, it is unavoidable to cause errors in these sampled data. This kind of data fluctuation cannot reflect the real water quality; hence it usually results in misjudgment of the water quality situation.

Wavelet decomposition is often used for de-noising analysis of sensor data [5,9,15]. Multiresolution analysis is one of the main functions of wavelet analysis, which was proposed by Meyer and Mallat [17]. It can decompose data sequences into different resolutions in space. A wavelet is defined as a function with a finite interval and a mean value of zero, and it is called the mother wavelet. There is also a scaling function called the father wavelet. Assuming that the signal S has a M samples, the J class Wavelet ID of the signal transform is expressed as [6]: $\begin{matrix} (1) & w_{j} = [w_{a J}, w_{d J}, \dots, w_{d 1}] \end{matrix}$ Wavelet coefficient $\begin{matrix} (2) & \begin{array}{l} w_{a J} = ⟨ s (n) \cdot φ_{J} (n) ⟩ \\ w_{d j} = ⟨ s (n) \cdot ψ_{j} (n) ⟩ \end{array} \end{matrix}$ Where: $\begin{matrix} (3) & \begin{array}{l} j = J, \dots, 1 \\ n = 1, \dots, M \end{array} \end{matrix}$ The father wavelet $φ (t)$ is equivalent to a low-pass filter, and the mother wavelet $ψ (t)$ is equivalent to a high-pass filter. Among them, the high-frequency data reflects random components caused by uncertainties in the data, such as irregular fluctuations and sensor errors. The low-frequency data represents the true trend of the data sequence.

The data can be reconstructed by inverse transform after using wavelet decomposition: $\begin{matrix} (4) & s (n) = w_{a J} φ_{J} (n) + \sum_{j = 1}^{J} w_{d j} ψ_{j} (n) \end{matrix}$ The system uses Daubechies wavelet, which implements a set of filters, can completely reconstruct the original signal [20]. DB5 is selected to decompose the data series into 4 levels based on the sampling frequency and application requirements of the sensors. In storm Bolt, it maintains a sliding window of 1024 data, and the date in the window will be decomposed into high and low frequent coefficients by wavelet in every 300 seconds, i.e. after every update of 100 data. The low-frequency coefficients can be reconstructed to obtain the change trend of the water quality in real-time. Meanwhile, the trend is sent to the Web page by WebSocket, where the users can easily recognize the future situation of the water quality.

4.3. LSTM model

Most current existing water quality monitoring systems provide only real time monitoring and data query functions. In many cases, it is essential for the system to have forecasting capability and provide users with the future water quality situations in advance.

A Long Short Term Memory Network (LSTM) is a type of Recurrent Neural Network (RNN), and it is specially designed to prevent the neural network output for a given input from either decaying or exploding as it cycles through the feedback loops [8]. The feedback loops make recurrent networks much better in pattern recognition than other neural networks. Memory of past input is critical for solving sequence learning tasks, and LSTM provides a better performance compared to other RNN architectures by alleviating the vanishing gradient problem. Due to their ability to learn long term dependencies, LSTMs are applicable to a number of sequence learning problems including language modeling and translation, acoustic modeling of speech, speech synthesis, speech recognition, audio and video data analysis, handwriting recognition and generation, sequence prediction, and protein secondary structure prediction, and it has an outstanding performance in time series prediction [26]. Because of the presence of fluctuation information in the original data of water quality, neural networks occupy a higher accuracy to predict trend of such kind of data [25].

In our system, the sensor sampling value is a time sequence with a fixed frequency, and its structure remains unchanged after wavelet decomposition and reconstruction. ${x_{1}, x_{2}, \dots, x_{t}}$ represents current trend data and uses LSTM to predict data ${x_{t + 1}, x_{t + 2}, \dots, x_{T}}$ . The LSTM uses two gates to control the contents of the unit state $c$ . One is the forget gate $f_{t}$ , which determines how much of the unit state at the previous moment $c_{t - 1}$ is retained until the current moment $c_{t}$ .The other is the memory gate, which determines how much of the network input $x_{t}$ is saved to the unit state $c_{t}$ at the current moment. Before LSTM prediction, training models need to be performed first, there are mainly three steps:

1) Forward calculating the output value of each neuron, which are the values of five vectors $f_{t}$ , $i_{t}$ , $c_{t}$ , $o_{t}$ , $h_{t}$ . Forget gate calculation formula goes as follows: $\begin{matrix} (5) & f_{t} = σ (W_{f} \cdot [h_{t - 1}, x_{t}] + b_{f}) \end{matrix}$

Where: $f_{t}$ is forget gate, σ is sigmoid function, $W_{f}$ is the weight matrix for the forget gate, $h_{t - 1}$ is the output in the time $t-1$ , $x_{t}$ is the input for the time t, $[h_{t - 1}, x_{t}]$ is the connection of the two vectors, and $b_{f}$ is the offset of the forget gate. Input gate calculation formula goes as follows: $\begin{matrix} (6) & i_{t} = σ (W_{i} \cdot [h_{t - 1}, x_{t}] + b_{i}) \end{matrix}$ where: $i_{t}$ is input gate, $W_{i}$ is the weight matrix of the input gate, and $b_{i}$ is the bias of the input gate.

The current unit state formula is: $\begin{matrix} (7) & c_{t} = f_{t} ⊙ c_{t - 1} + i_{t} ⊙ {\tilde{c}}_{t - 1} \end{matrix}$ where: ⊙ means multiplies by element; ${\tilde{c}}_{t - 1}$ is the current unit state. $\begin{matrix} (8) & {\tilde{c}}_{t} = tanh (W_{c} \cdot [h_{t - 1}, x_{t}] + b_{c}) \end{matrix}$ where: $W_{c}$ is the weight matrix for current states and $b_{c}$ is the bias term of the current states.

The computing formula at the current unit state is expressed by: $\begin{matrix} (9) & o_{t} = σ (W_{o} \cdot [h_{t - 1}, x_{t}] + b_{o}) \end{matrix}$ where: $o_{t}$ is the input gate, $W_{o}$ is the weight matrix for the output gate, $b_{o}$ is the bias term of the out gate.

The final output of the LSTM is determined by the output gate and unit states: $\begin{matrix} (10) & h_{t - 1} = o_{t} ⊙ tanh (c_{t}) \end{matrix}$

2) Reversely calculating the error term of each neuron. The back propagation of the LSTM error term includes two directions: one is the backward propagation along time from the current time t to calculate the error term for each moment; another is to propagate the error term to the upper layer.

3) Calculate the gradient of each weight based on the corresponding error term. In our system, we set 100 data, i.e. 5-minute data in the sliding window to predict the values in the next 5 minutes. One hidden layer is set for the neural network in our system. For the wavelet decomposed 1024 data sequences in the sliding windows, the first 896 values are used for model training and the last 128 are used for testing. The learning rate of each layer of the network node is 0.2, and Mean Square Error (MSE) and $R^{2}$ is used as the loss and accuracy functions respectively for evaluation of the model training and validation results. The MSE and $R^{2}$ formulas are: $\begin{array}{l} (11) & MSE = \frac{\sum_{i = 1}^{m} {(y_{i} - {\hat{y}}_{i})}^{2}}{m} \\ (12) & R^{2} = 1 - \frac{\sum_{i} {({\hat{y}}_{i} - y_{i})}^{2}}{\sum_{i} {({\bar{y}}_{i} - y_{i})}^{2}} \end{array}$ where ${\hat{y}}_{i}$ is the real value, $y_{i}$ is the average value and ${\bar{y}}_{i}$ is the predicted value.

It is online training, thus the value of batch_size parameter is set as 1. The trained model predicts one step ahead based on the current trend obtained by wavelet decomposition, and then the predicted value is used as the last value of the trend sequence to predict next one step ahead value. This process iterates 100 times, and consequently the monitoring indicators in the next five minutes are predicted.

4.4. Data storage

Besides analysis, sensor data also needs to be stored in the hard disk for historical data query. In the system, HBase is selected as the data storage database due to its nature of storing high sampling frequency with large data storage size.

HBase is a distributed column-oriented No-SQL database built on the top of Hadoop file system, and it is a data model that is similar to Googled big table designed to provide quick random access to huge amounts of structured data [2]. HBase is a column- oriented database and the tables in it are sorted by row. The table schema defines only column families, which are the key value pairs. A table have multiple column families and each column family can have any number of columns. Subsequent column values are stored contiguously in the disk, and each cell value of the table has a timestamp.

HBase can store large data sets on top of HDFS file storage, and it will aggregate and analyze billions of rows present in its tables. Operations such as data reading and processing will take less time as compared to traditional relational models. In addition, HBase is used extensively for online analytical operations, and the database also can be shared in HBase. These features are helpful for the sensor data from different monitor areas to be stored into different tables.

Mehmood and Anees made an analysis on the CPU and memory usage for real-time semi-stream join processing through two types of tests, unstructured and structured data streams based on synthetic and real datasets. The results show that memory usage and execution time remains consistent for a given specification irrespective of the nature of data streams (unstructured or structured), even when incoming semi-streams are growing [18].

Since the RowKey is critical in HBase, it should be designed based on the data access patterns of the business requirement and the data storage characteristics of HBase. RowKey is a binary code in HBase, and it can be any character with the length of no more than 64k bytes. According to the characteristics of the historical data query, it will be more convenient to use the date and time as the RowKey. However, the continuous time values will cause the data to be concentrated in a region of the cluster, which causes hotspot issues and degrades the cluster performance and even makes it crash.

In order to solve these problems, a table is set up based on the monitored water areas and monitored points in the system.

The RowKey format of is: $\begin{array}{l} Sensor number (2 bytes) + date (8 bytes) \\ + time (6 bytes) \end{array}$ This design makes the RowKey length as an integer times of 8 bytes, which fully utilizes the 8-bit alignment of the 64-bit operating system memory to achieve the best performance of the system. In addition, RowKey starts with the sensor number, which allows the data to be stored relatively even in each region of the cluster to reach the optimal cluster performance.

5. Result analysis

5.1. System environment

In our system, DELL PowerEdge R710 server is used to build clusters, and the server hardware is configured as 96G memory, 24 core X5650*2CPU, 1800G hard disk. The server is virtualized with three virtual machines, and each virtual machine is allocated 24G memory, 24 core CPU and 500G hard disk. Zookeeper is employed to provide application coordination services, and a HB cluster, a Kafka cluster and a storm platform are built in the virtual machines. The development tool is JDK1.8, Apache Storm version is 1.1.0, HBASE version is 1.3.1, wavelet decomposition uses open source Jwavel.0.3, LSTM uses TensorFlow 1.4.0. The sensors produced by Wuhan Fuyuanfeike Company are used to minitor the indecies of pH, DO and COD. The wireless gateway uses the F2614 CDMA 4G wireless DTU of Xiamen Sixin Company. The selected hardware is shown in Fig. 4.

Since one monitoring point in the monitored water area has been set so far, one TOPIC in Kafka, named xinAn_0001 was created. In order to simulate multiple monitoring points, three copies of the monitored point information were copied and three partitions were created by the receiving software. In the experiments, multiple monitoring points were simulated by software.

Fig. 4.

Sensors and wireless DTU.

5.2. Experimental data

Statistical analysis result of pH, DO and Water temperature data taken from one sliding window in Storm Bolt is shown in Table 1. For each index, 895 of 1024 data set is used for training and the rest 128 is used for testing. The analysis result reveals that amplitude of the three data sets is comparatively stable in general. Whereas smaller standard deviation of the three testing data sets indicates that the values of testing data tend to be much closer to the mean, which might cause prediction values is larger than the observed ones.

Table 1
Statistical analysis of the data

Index Data set count mean std min 25% 50% 75% max

pH Train 895 8.6877 0.0204 8.6435 8.6698 8.6481 8.706 8.7267

Test 128 8.6669 0.009 8.6435 8.6622 8.6684 8.6719 8.6863

Total 1024 8.6851 0.0205 8.6435 8.6687 8.681 8.7028 8.7267

DO (mg/L) Train 895 1.4856 0.0068 1.4724 1.4794 1.4873 1.491 1.497

Test 128 1.4793 0.0056 1.4729 1.4749 1.477 1.4835 1.4936

Total 1024 1.4848 0.007 1.4724 1.4776 1.4861 1.4904 1.497

Water temperature (°C) Train 895 25.262 0.0626 25.1536 25.2242 25.2427 25.2826 25.4334

Test 128 25.3711 0.0303 25.298 25.3534 25.3749 25.3934 25.4334

Total 1024 25.2756 0.0697 25.1536 25.2273 25.2488 25.3411 25.4334

Index	Data set	count	mean	std	min	25%	50%	75%	max
pH	Train	895	8.6877	0.0204	8.6435	8.6698	8.6481	8.706	8.7267
Test	128	8.6669	0.009	8.6435	8.6622	8.6684	8.6719	8.6863
Total	1024	8.6851	0.0205	8.6435	8.6687	8.681	8.7028	8.7267
DO (mg/L)	Train	895	1.4856	0.0068	1.4724	1.4794	1.4873	1.491	1.497
Test	128	1.4793	0.0056	1.4729	1.4749	1.477	1.4835	1.4936
Total	1024	1.4848	0.007	1.4724	1.4776	1.4861	1.4904	1.497
Water temperature (°C)	Train	895	25.262	0.0626	25.1536	25.2242	25.2427	25.2826	25.4334
Test	128	25.3711	0.0303	25.298	25.3534	25.3749	25.3934	25.4334
Total	1024	25.2756	0.0697	25.1536	25.2273	25.2488	25.3411	25.4334

5.3. Wavelet decomposition results

Figure 5 displays the comparison results between the original series of pH, DO and temperature and their wavelet decomposed approximation series (trend). The results reveal that the approximation series effectively filter the fluctuation of the original data and they more obviously display the trend of original series.

Fig. 5.

Original data and trend line, (a) pH, (b) DO, (c) water temperature.

5.4. Training and validation results

During the training process, 10% of 895 training data is further used for model validation in order to keep training from overfitting. Table 2 shows the training results after 100 epochs, and the training results show that MSE loss of training fitting and validation are both smaller (0.0008~0.0022) while $R^{2}$ are very higher (0.994~0.998). This result confirms that the model has an excellent performance to fit the approximation lines of the three indices with a high accuracy. Figure 6 illustrates the MSE loss of the model training and validation of 100 epochs, where it reveals that MSE losses of the three indices all decrease rapidly and it has reached a very lower loss after 20 epochs for pH (Fig. 6a), over 40 epochs for DO and Water temperature (Fig. 6b, 6c).

Fig. 6.

Mean Squared Error (MSE) loss of model training and validation of 100 epochs, (a) pH, (b) DO, (c) water temperature.

5.5. Testing result

The testing results show that prediction MSE losses of the three indices are very low (0.0038~0.0009) with higher good of fitness ( $R^{2} ⩾ 9.5$ ), indicating that model has high prediction ability (Table 2). As an example, Fig. 7 illustrates a comparison results between the sensor measured pH values and the predicted values. The predicted values of three indices are basically consistent with its observed values, indicating that the proposed prediction algorithm is effective. This visual comparison further illustrates the good prediction performances of the model.

Table 2
Results of model training, validation and testing

Index Training Validation Testing

MSE $R^{2}$ MSE $R^{2}$ MSE $R^{2}$

pH 0.0008 0.9966 0.0008 0.9976 0.0009 0.9519

DO 0.0022 0.9937 0.0024 0.9944 0.0038 0.9534

Temperature 0.0010 0.9962 0.0007 0.9981 0.0005 0.9668

Index	Training	Validation	Testing
pH	0.0008	0.9966	0.0008	0.9976	0.0009	0.9519
DO	0.0022	0.9937	0.0024	0.9944	0.0038	0.9534
Temperature	0.0010	0.9962	0.0007	0.9981	0.0005	0.9668

5.6. Processing capability

For the processing capabilities of the Storm platform, the Complete latency is the whole processing time of the tuple and the tuple tree, while the Execute latency is the time spent by Bolt to process the tuples. Capacity is equal to the numbers of messages, processed by the Bolt calling the Execute method, multiplying by the average time of the message, and then divided by the time interval of the message. Capacity reflects the Bolt’s idleness, where the greater value indicates that the data transmission speed of Spout is faster, the but Bolt has a much heavier burden. Table 3 shows the comparison between the three indicators in three and ten monitoring points. It reveals that the number of data processing delays is raised with increase of the node numbers, whereas it still fully satisfies the business requirements. In addition, the change of Capacity is small, indicating that the Storm platform is robust enough. Figures 8, 9, and 10 display the server hardware loads of 10 monitoring points. The load statistic confirms that the system resource load is basically at a stable level. Consequently, it indicates that, with the support of appropriate hardware resources, Storm platform will be more robust in processing real-time data streams. In Fig. 8, the Y axis is the number of processes and the X axis is the time. From Fig. 8 we can know that seldom process costs long time and most of the processes cost mid or short time. In Fig. 9 and Fig. 10, the X axises are time, and Y axis is memory cost and CPU cost respectively, they show that the resource costs are reasonable.

Table 3
Performance of storm with different loads

Monitor points number Complete latency (ms) Execute latency (ms) Capacity

3 34.142 3.436 0.075

10 68.05 15.397 0.116

Monitor points number	Complete latency (ms)	Execute latency (ms)	Capacity
3	34.142	3.436	0.075
10	68.05	15.397	0.116

Fig. 7.

Comparison of observation and prediction of pH.

Fig. 8.

System load of last 24 hours.

Fig. 9.

Memory load of last 24 hours.

Fig. 10.

CPU load of last 24 hours.

6. Conclusions and future works

At present, a big data water quality monitoring and processing platform framework has been completed, and the advantages of the system proposed integrate IoT devices, Apache storm and Wavelet transformation and deep learning modelling approach. It solves the deficiencies of traditional water monitoring system, which can be used to process high frequency real time data, monitoring and predicting the situations of water quality and giving an early warning on the predicted problems. The main experimental results indicate that the LSTM based deep learning model in the system has a good performance to model and predict water quality with high accuracy, and the big data platform has robust capacity to process the vast amount of high frequent sensor data. However, the following work needs to do further in the future: 1) Multi-feature water quality prediction. The platform is currently monitoring three water quality indicators in one monitored point and provides early warning based on prediction results. However, more advanced application requires comprehensively forecasting even more multiple water quality indicators in multiple monitored points. Hence it requires integrating the multi-features to train the model, and then making classification predictions of the multiple water quality indicators. 2) Dynamic temporal granularity forecast. The current forecasting time granularity is in seconds. Such a designed model is helpful in forecasting sudden events, while it is incapable of making a mid-to-long-term forecast of the water quality. The dynamic time granularity is adjusted according to the user’s demand to predict the water quality in the next few days or months. Such demand needs to fuse the historical data and extract the data features for model training, validation and testing according to the time granularity.

Footnotes

Acknowledgements

This work was supported by the Scientific Bureau of Yantai, China under the Science and Technology Plan Project “Green and Efficient Smart Agricultural Management Platform Based on New Generation of Information Technology” (2018YT06130844).

References

K.S.

Adu-Manu,

Tapparello,

Heinzelman,

F.A.

Katsriku and

J.D.

Abdulai, Water quality monitoring using wireless sensor networks: Current trends and future research directions, ACM Transactions on Sensor Networks 13 (2017), 1–41. doi:10.1145/3005719.

Apache HBase Tutorial [EB/OL], Apache Software Foundation, https://hbase.apache.org/.

Apache Storm Prototype [EB/OL], Apache Software Foundation, http://storm.apache.org/.

Apache Storm Trident Tutorial [EB/OL], Apache Software Foundation, http://storm.apache.org/releases/current/Trident-tutorial.html.

A.L.L.

Aquino,

R.A.

Rabelo and

E.F.

Wanner, A wavelet-based sampling algorithm for wireless sensor networks applications, in: Proceedings of the 2010 ACM Symposium on Applied Computing, ACM, 2010, pp. 1604–1608. doi:10.1145/1774088.1774433.

Daubechies, Ten lectures on wavelets, The Journal of the Acoustical Society of America 93 (1992), 697. doi:10.1121/1.406784.

M.A.

Guillén-Navarro,

Martínez,

Llanes-Castro,

Bueno-Crespo and

J.M.

Cecilia, A deep learning model to predict lower temperatures in agriculture, Journal of Ambient Intelligence and Smart Environments, IOS Press 12 (2020), 1–14. doi:10.3233/AIS-200546.

Hochreiter and

Schmidhuber, Long short-term memory, Neural computation 9 (1997), 1735–1780. doi:10.1162/neco.1997.9.8.1735.

Hu,

Meng,

Celentano and

Yang, Robust medical ad hoc sensor networks (MASN) with wavelet-based ECG data mining, Ad Hoc Networks 6 (2008), 986–1012. doi:10.1016/j.adhoc.2007.09.002.

10.

Hu,

Sun,

Wang,

X.H.

Fei and

Zhao, A stream processing system for multisource heterogeneous sensor data, Journal of Sensors 2016 (2016), 1–8. doi:10.1155/2016/4287834.

11.

Huang,

Meng,

Zhang and

Zhang, In-memory parallel processing of massive remotely sensed data using an apache spark on hadoop YARN model, IEEE Journal of Selected Topics in Applied Earth Observations and Remote Sensing 10 (2017), 3–19. doi:10.1109/JSTARS.2016.2547020.

12.

Jardak,

Riihijärvi,

Oldewurtel and

Mähönen, Parallel processing of data from very large-scale wireless sensor networks, in: Proceedings of the 19th ACM International Symposium on High Performance Distributed Computing, Association for Computing Machinery, ACM, 2010, pp. 787–794. doi:10.1145/1851476.1851590.

13.

Krish,

Anusuya,

Kumar and

L.H.

Son, Real-time water quality monitoring using Internet of things in SCADA, Environmental Monitoring and Assessment 190 (2018), 556. doi:10.1007/s10661-018-6914-x.

14.

J.J.

Lei,

Meyer and

R.D.

Ryan, Wavelets: Algorithms & applications, Mathematics of Computation, JSTOR 63 (1994), 822. doi:10.2307/2153305.

15.

S.-Y.

Li,

Lin,

G.-H.

Yang,

P.-P.

Tian,

Y.Z.H.

Wang,

S.-B.

Yu and

Y.-J.

Ji, Ground-airborne electromagnetic signals de-noising using a combined wavelet transform algorithm, Chinese Journal of Geophysics 56 (2013), 3145–3152, (in Chinese). doi:10.6038/cjg20130927.

16.

Madhavireddy and

Koteswarrao, Smart water quality monitoring system using IoT technology, International Journal of Engineering and Technology(UAE) 7 (2018), 636–639. doi:10.14419/ijet.v7i4.36.24214.

17.

S.G.

Mallat, Multiresolution Representations and Wavelets, Ph.D. Dissertation, University of Pennsylvania, 1988.

18.

Mehmood and

Anees, Performance analysis of not only SQL semi-stream join using MongoDB for real-time data warehousing, IEEE Access 7 (2019), 134215–134225. doi:10.1109/ACCESS.2019.2941925.

19.

Olap, A Study on Wikipedia Clickstream Data Analysis with Apache Storm, Northeastern University, Master Thesis, 2018. doi:10.13140/RG.2.2.11064.39688.

20.

Pule,

Yahya and

Chuma, Wireless sensor networks: A survey on monitoring water quality, Journal of Applied Research and Technology 15 (2017), 562–570. doi:10.1016/j.jart.2017.07.004.

21.

T.X.

Shu,

Xia,

J.H.

Chen and

Silva, An energy efficient adaptive sampling algorithm in a sensor network for auto mated water quality monitoring, Sensors 17 (2017), 2551. doi:10.3390/s17112551.

22.

M.A.

Talhaoui, Real-time data stream processing – challenges and perspectives, International Journal of Computer Science Issues 14 (2018), 6–12. doi:10.20943/01201705.612.

23.

Vijayakumar and

Ramya, The real time monitoring of water quality in IoT environment, in: 2015 International Conference on Circuits, Power and Computing Technologies [ICCPCT-2015], IEEE, 2015, pp. 1–4. doi:10.1109/ICIIECS.2015.7193080.

24.

Wang and

A.L.

Yalcin, Health and wellness monitoring using ambient sensor networks, Journal of Ambient Intelligence and Smart Environments, IOS Press 12 (2020), 139–151. doi:10.3233/AIS-200553.

25.

S.K.

Wei,

Yang,

J.X.

Song,

Abbaspour and

Z.X.

Xu, A wavelet-neural network hybrid modeling approach for estimating and predicting river monthly flows, Hydrological Sciences Journal 58 (2013), 374–389. doi:10.1080/02626667.2012.754102.

26.

H.-M.

Yang,

Z.-S.

Pan and

Tao, Robust and adaptive online time series prediction with long short-term memory, Computational Intelligence and Neuroscience 2017 (2017), 1–9. doi:10.1155/2017/9478952.

27.

L.C.

Yann,

Bengio and

Hinton, Deep learning, Nature 521 (2015), 436–444. doi:10.1038/nature14539.

28.

Zennaro,

Floros,

Dogan,

Sun,

Z.C.

Cao,

Huang,

Bahader,

Ntareme and

Bagula, On the design of a water quality wireless sensor network (WQWSN): An application toWater quality monitoring in Malawi, in: 2009 International Conference on Parallel Processing Workshops, IEEE, 2009, pp. 330–336. doi:10.1109/ICPPW.2009.57.

29.

L.J.

Zhou,

N.C.

Chen and

Z.Q.

Chen, Efficient streaming mass spatio-temporal vehicle data access in urban sensor networks based on apache storm, Sensors 17 (2017), 815. doi:10.3390/s17040815.

Analysis and prediction of big stream data in real-time water quality monitoring system

Abstract

Keywords

1. Introduction

2. Related works

3. Sytem design

4.1. Serial data parallelization

4.3. LSTM model

4.4. Data storage

5. Result analysis

5.1. System environment

Table 2 Results of model training, validation and testing Index Training Validation Testing MSE R 2 MSE R 2 MSE R 2 pH 0.0008 0.9966 0.0008 0.9976 0.0009 0.9519 DO 0.0022 0.9937 0.0024 0.9944 0.0038 0.9534 Temperature 0.0010 0.9962 0.0007 0.9981 0.0005 0.9668

Table 3 Performance of storm with different loads Monitor points number Complete latency (ms) Execute latency (ms) Capacity 3 34.142 3.436 0.075 10 68.05 15.397 0.116

Footnotes

Acknowledgements

References

Table 2
Results of model training, validation and testing

Index Training Validation Testing

MSE $R^{2}$ MSE $R^{2}$ MSE $R^{2}$

pH 0.0008 0.9966 0.0008 0.9976 0.0009 0.9519

DO 0.0022 0.9937 0.0024 0.9944 0.0038 0.9534

Temperature 0.0010 0.9962 0.0007 0.9981 0.0005 0.9668

Table 3
Performance of storm with different loads

Monitor points number Complete latency (ms) Execute latency (ms) Capacity

3 34.142 3.436 0.075

10 68.05 15.397 0.116