Kafka and MQTT Application > Source Code Overview

Data Viewer

From this point, connecting and configuring the GridDB instance will be identified as connecting a single sensor; just make sure to add the static classes from the GridDBSinkTask.java program into this file.

The main difference in this version is that we will be querying from multiple containers and using multi-get and multi-put functionality to group sensor readings of the same type into one set of rows so we that may gather statistics such as the average reading or the max reading.

Using Multi-Get Queries

From there, using the sensor-id of each record all the individual sensors’ data can be queried, displayed, and statistics regarding average, and max values can be gathered using a multiget query.

Creating the multi-get query(DataViewer.java)

// Create an entry in a predicate that will correspond to an individual 
// sensor-id  as a key 
// and a date-range (type RowKeyPredicate<Date>) query as the value
private static void setMultiGetMaps(RowSet<SensorType> rowSet, RowKeyPredicate<Date> dateKey) throws GSException {
	while(rowSet.hasNext()){
		SensorType rowType = rowSet.next();
		typeMap.get(rowType.type).put(rowType.id,dateKey);
	}
}
// (snip)
// Get all records from the sensor-type Collection
String tql = "SELECT * FROM " + "SENSOR_TYPE_NAME"; 
// Create and fetch results for the query
Query<SensorType> query = typeContainer.query(tql,SensorType.class);
RowSet<SensorType> rowSet = query.fetch(); 
setMultiGetMaps(rowSet, dateKeys);

Execute multiget query(DataViewer.java)

typeMap will have a map of maps each containing sensor-id, dateKey query as key-value pairs. The outer map consists of sensor-types, predicate-map key-value pairs.

Map<String, List<Row>> multiGetResults = store.multiGet(entry.getValue());
HashMap<String,Double> typeAveragesMap = new HashMap<>();
for(Entry<String, List<Row>> getResult : multiGetResults.entrySet()){
	String id = getResult.getKey(); // Get the sensor-id
	HashMap<String,Double< sensorAverages = getSensorData(type,id,start,end);
	if(typeAveragesMap.isEmpty())
		typeAveragesMap = sensorAverages;
	else {
		for(Entry<String,Double> pair : sensorAverages.entrySet()){
			String key = pair.getKey();
			Double value = pair.getValue();
			typeAveragesMap.put(key,typeAveragesMap.get(key) + value);
		}
	}
}

Each multiget result will return a HashMap of sensor-id, row-list pairs which is stored in the variable multiGetResults. This HashMap has the sensor-id as the key and all the records it corresponds to as the value which is a list of type Row. Note that every sensor-id is also the container name of a time series container where we get aggregation information. We obtain and output all these statistics through the use of a function called getSensorData(sensor-id,type,starting-date,ending-date). The only other information we need is just the type of sensor, which is the key of the outer HashMap, typeMap.

Collecting Averages From Each Sensor

Once getSensorData(sensor-id,type,starting-date,ending-date) is executed, it returns a HashMap of the average values of each field times the number of records that is stored in the container that is named by its sensor-id parameter. This HashMap of an individual sensor is stored in the variable sensorAverages. We also want to have another HashMap in the variable typeAveragesMap that we can use to store the averages of all the fields or metrics a type of sensor can measure. An example is a light sensor can measure light and sound readings, so we would want to keep a running count of all the averages of every individual sensor and the amount of records it has.

We can eventually compute an average for all the light sensors, or for all the watt sensors, etc. We do this by iterating over every key in typesAveragesMap and adding its value by the matching key in sensorAverages. An example is adding the “light” value of typeAveragesMap HashMap by the “light” value of the sensorAverages HashMap.

typeAveragesMap.put(key,typeAveragesMap.get(key) + value);

Once all the averages from every sensor of a certain type have been collected, we can use the typeAveragesMap to calculate and display the overall averages. This is achieved by using the function displayAverages().

Aggregation Methods

GridDB provides a variety of built in API functions that can be used to aggregate data and collect statistics. Aggregate methods return Aggregate Results usually in the form of numbers rather than a set of rows from the container. For any container with numeric-type columns aggregation queries can be made on those columns.

Aggregating queries on all columns(DataViewer.java : getRowCounts)

for(int i = 1; i < columnCount; i++){ //Skip over the timestamp column
	String fieldId = containerInfo.getColumnInfo(i).getName(); // Get column name
	AggregationResult minSet = timeSeries.aggregate(start,end,fieldId,Aggregation.MINIMUM); // Get minimum
	AggregationResult averageSet =  timeSeries.aggregate(start,end,fieldId,Aggregation.WEIGHTED_AVERAGE); //time-weighted avg
	AggregationResult maxSet = timeSeries.aggregate(start,end,fieldId,Aggregation.MAXIMUM); // Get maximum
	AggregationResult stdDev = timeSeries.aggregate(start,end,fieldId,Aggregation.STANDARD_DEVIATION);

We can iterate through every column except the first timestamp to obtain its name and aggregation results in the form of the weighted average value (accounts for variable timespans), minimum values, maximum values, and standard deviation for each column name.

Data for a Single Sensor

To display data for a single sensor device you can can use getSensorData() and use the id and type of device as parameters.

So to get the statistical information on light Sensor “A_10”

getSensorData(“light”, “A_10”, starting_date, ending_date);

Output:

Sensor Readings from light Sensor with ID: A_10
Avg: light 57.472439 Min: light 1.902972 Max: light 95.789162 Std.Dev: light output is: 32.465236
Avg: sound 69.132693 Min: sound 10.343853 Max: sound 89.096977 Std.Dev: sound output is: 24.447340

To get statistical information on all columns of watt Sensor “C_1”:

getSensorData(“watts”,”C_1”,starting_date,ending_date);

Output:

Sensor Readings from watts Sensor with ID: C_1
Avg: watts 34.060980 Min: watts 2.378983 Max: watts 99.227776 Std.Dev: watts output is: 26.922309	 
Avg: heat 71.681186 Min: heat 3.175400 Max: heat 100.148102 Std.Dev: heat output is: 30.178766

To get statistical information on all columns of volt sensor “B_4”

Output:

Sensor Readings from volts Sensor with ID: B_4
Avg: volts 24.282921 Min: volts 7.597842 Max: volts 99.006050 Std.Dev: volts output is: 29.264624	Avg: amps 27.462714 Min: amps 9.404010 Max: amps 99.905853 Std.Dev: amps output is: 34.040586

Displaying All Averages

To obtain and output the overall average for a certain sensor type, such as the average heat output and watt output for all watt sensors, we use the function displayAverages().

private static void displayAverages(HashMap<String,Double> sensorAggregationMap){
	Double count = sensorAggregationMap.get("count");
	for(Entry<String, Double> averagePair : sensorAggregationMap.entrySet()){
		if(!(averagePair.getKey().equals("count"))){
			String displayedAverages = String.format("Avg %s %f",averagePair.getKey(), averagePair.getValue() / count );
			System.out.print(displayedAverages + "\t");
		}
	}
	System.out.println();
}

This function takes the HashMap from the earlier variable typeAverages. The keys of typeAverages are the names of the fields or metrics a sensor can measure such as how an electricity sensor can measure voltage and amperage. The values are the sums of the averages of all the sensors of a certain type. There is also an additional key, count, which has the number of records a type of sensor has in total. To calculate the overall average, all that is needed is to divide each sum of averages by that count value (number of all records in total). Once we have that value we can output the results.

Example Output for All Light Sensors:

All Light Sensors:
Avg light: 59.79225 Avg Sound: 53.225707

Source Code

Complete source code used for the GridDB Kafka and MQTT Application can be downloaded from the following:

Download: griddb-kafka-mqtt-application.tar.gz