Red Hat’s JBoss Data Grid 7 is an open source, distributed, in-memory key/value data store based on the Infinispan open source software project. Whether deployed in client/server mode or embedded in a Java Virtual Machine, it is built to be elastic, high performance, highly available and to scale linearly. Refer to the JDG 7 documentation for further details.

JBoss Data Grid is accessible for both Java and non-Java clients. Using JBoss Data Grid, data is distributed and replicated across a manageable cluster of nodes, optionally written to disk and easily accessible using the REST, Memcached and Hot Rod protocols, or directly in process through a traditional Java Map API.

Red Hat JBoss Data Grid offers two usage modes:

The use of a dedicated JVM for JBoss Data Grid allows for appropriate tuning and configuration for JDG versus Spark and client applications. In particular, handling the memory requirements of both Apache Spark and JBoss Data Grid in a single JVM can be difficult. For this and other reasons, Apache Spark integration support is only provided in Client-Server mode and this reference architecture sets up JBoss Data Grid accordingly.

JDG 7.0 introduces Resilient Distributed Dataset (RDD) and Discretized Stream (DStream) integration with Apache Spark version 1.6.0. This enables you to use JDG as a highly scalable, high-performance data source for Apache Spark, executing Spark and Spark Streaming operations on data stored in JDG. Refer to the JDG 7 documentation for further details.

Installing Apache Spark is very simple and mainly involves extracting the downloaded archive file on each node.

# tar xvf spark-1.6.0-bin-hadoop2.6.tgz

JBoss Data Grid 7 does not require any installation steps. The archive file simply needs to be extracted after the download. This reference architecture requires installation of JBoss Data Grid 7.0.0 Server on each node.

# unzip jboss-datagrid-7.0.0-server.zip -d /opt/

Various other types of configuration may be required for UDP and TCP communication. For example, Linux operating systems typically have a low maximum socket buffer size configured, which is lower than the default cluster JGroups buffer size. It may be important to correct any such warnings observed in the JDG logs. For more information, please follow the Administration and Configuration Guide for JDG 7

This reference architecture installs and configures a three-node cluster on separate machines. The names node1, node2 and node3 are used in this paper to refer to both the machines and the JDG 7 nodes on them.

Figure 3.1. Deployment Clusters

# /opt/jboss-datagrid-7.0.0-server/bin/add-user.sh admin password1!

# /opt/jboss-datagrid-7.0.0-server/bin/add-user.sh

On node1, execute the following scripts to add two new JDG 7 distributed caches, sensor-data and sensor-avg-data. These 2 caches will be used by the sample IoT sensor application.

# /opt/jboss-datagrid-7.0.0-server/bin/cli.sh # embed-host-controller --domain-config=domain.xml --host-config=host.xml --std-out=echo # /profile=clustered/subsystem=datagrid-infinispan/cache-container=clustered/configurations=CONFIGURATIONS/distributed-cache-configuration=sensor-data:add(start=EAGER,template=false,mode=SYNC) # /profile=clustered/subsystem=datagrid-infinispan/cache-container=clustered/distributed-cache=sensor-data:add(configuration=sensor-data) # /profile=clustered/subsystem=datagrid-infinispan/cache-container=clustered/configurations=CONFIGURATIONS/distributed-cache-configuration=sensor-avg-data:add(start=EAGER,template=false,mode=SYNC) # /profile=clustered/subsystem=datagrid-infinispan/cache-container=clustered/distributed-cache=sensor-avg-data:add(configuration=sensor-avg-data) 

1) Update /opt/jboss-datagrid-7.0.0-server/domain/configuration/host-slave.xml on both node2 and node3, so these 2 nodes can form a JDG cluster with node1.

Update the first line for node2, adding host name.

Update the first line for node3, adding host name.

2) Update node2 and node3’s host-slave.xml, change server-identities value from default sample value to this new value, which is the encrypted password value of the node user from last section.

3) Update the server name for node1 in host.xml, by deleting the server-two tag.

Update the server name to server-two for node2 in host-slave.xml.

Update the server name to server-three- for node3 in _host-slave.xml.

After these change, the cluster will have 3 members, server-one on node1, server-two on node2 and server-three on node3.

Log on to the three machines where JDG 7 is installed and navigate to the bin directory:

# cd /opt/jboss-datagrid-7.0.0-server/bin

To start the first node

# ./domain.sh -bmanagement=10.19.137.34 -b=10.19.137.34

To start the second node

# ./domain.sh -b=10.19.137.35 -bprivate=10.19.137.35 --master-address=10.19.137.34 --host-config=host-slave.xml

To start the third node

# ./domain.sh -b=10.19.137.36 -bprivate=10.19.137.36 --master-address=10.19.137.34 --host-config=host-slave.xml

To stop the sensor applications, press ctrl-c or use "kill -9 PID" to stop the process.

Apache Spark currently supports three types of cluster managers:

This reference architecture uses the standalone cluster mode.

# cd /opt/spark-1.6.0-bin-hadoop2.6/sbin # ./start-master.sh --webui-port 9080 -h 10.19.137.34 # ./start-slave.sh spark://10.19.137.34:7077 --webui-port 9081 

# cd /opt/spark-1.6.0-bin-hadoop2.6/sbin # ./start-slave.sh spark://10.19.137.34:7077 --webui-port 9081 

# cd /opt/spark-1.6.0-bin-hadoop2.6/sbin # ./start-slave.sh spark://10.19.137.34:7077 --webui-port 9081 

On Node1, stop both master and worker.

# cd /opt/spark-1.6.0-bin-hadoop2.6/sbin # ./stop-slave.sh # ./stop-master.sh 

On Node2 and Node3, only need to stop the worker.

# cd /opt/spark-1.6.0-bin-hadoop2.6/sbin # ./stop-slave.sh 

To stop the Sensor applications, press ctrl-c or use "kill -9 PID" to stop the process. Otherwise, all 3 applications are set up to run for 24 hours.

SensorDataGenerator is the inner class which extends TimerTask; the main purpose for this class is to generate the data which simulates real world IoT data:

	private static class SensorDataGenerator extends TimerTask {

To simulate the data of IoT sensors from all over the world, this class will generate temperature and humidity data, and set these values for random shipments (shipment id from 0 to 999).

The key for ShipmentData is the shipment field; the value of it won’t change during the whole trip, just like in the real world. Each shipment could contain different items, which are identified by SKU numbers; it could be one or multiple items, and during the trip, the SKU number won’t change as well.

For demonstration purposes and in our example, if the random shipment id is between 0 to 250, the SKU number will be assigned as meat shipment. If the random shipment id is between 250 to 500, the SKU number will be assigned as metal shipment. if the random shipment id is between 500 to 750, the SKU number will be assigned as animal shipment. For shipment id between 750 to 1000, it’s a concatenated string of metal and animal, to demo a shipment that has both animal and metal products in it.

			String shipment = "shipment" + i;
			String sku = "sku_";
			double temperature = RANDOM.nextDouble() * TEMP_MAX;
			double humidity = RANDOM.nextDouble() * HUMIDITY_MAX;
			// setup the fake data, that the first batch of sku# is for meat shipment, next batch is metal, next batch is animal, last batch for combined shipment
			if (i < 250) {
				sku = sku + "meat" + i;
				temperature = RANDOM.nextDouble() * 30;
			}
			if (i >= 250 & i < 500) {
				sku = sku + "metal" + i;
				humidity = RANDOM.nextDouble() * 40;
			}
			if (i >= 500 & i < 750) {
				sku = sku + "animal" + i;
				temperature = RANDOM.nextDouble() * 45;
				humidity = RANDOM.nextDouble() * 75;
			}
			if (i >= 750 & i < 1000) {
				sku = sku + "animal" + i + "," + sku + "metal" + i;
			}

The main method of the TestSensor class will set up the Hot Rod Java Client to JDG cluster and connect to the sensor-data cache:

		// Configure remote cache
		ConfigurationBuilder builder = new ConfigurationBuilder();
		String DATAGRID_IP = args[0];
		builder.addServer().host(DATAGRID_IP).port(ConfigurationProperties.DEFAULT_HOTROD_PORT);
		RemoteCacheManager cacheManager = new RemoteCacheManager(builder.build());
		RemoteCache<String, ShipmentData> cache = cacheManager.getCache(CACHE_NAME);

Then it will start the timer and assign the task, which generates random sensor data.

		// Insert some sensor data into the cache
		TimerTask randomSensorData = new SensorDataGenerator(cache);
		Timer timer = new Timer(true);
		timer.schedule(randomSensorData, 0, GENERATE_PERIOD);

e.g. Below is sample of random data output.

First declare both JDG caches that will be used in this class:

  val inputDataGridCache = "sensor-data"
  val outputDataGridCache = "sensor-avg-data"

Get the JDG 7 server address from the first argument provided when starting the server, it can be the full cluster address like this: 10.19.137.34:11222;10.19.137.35:11222;10.19.137.36:11222

    val dataGridServer = args(0)

Next, initialize settings for spark stream and checkpoint path.

    // Initialize the Spark streaming context, with a batch duration of 1 second.
    val sparkConf = new SparkConf().setAppName("SparkSensor")
    val ssc = new StreamingContext(sparkConf, Seconds(1))
    // set up checkpoint to store state information
    ssc.checkpoint("/mnt/shared/sharefs/spark-sensor")

Then create the Spark stream.

    // configure the connection to the DataGrid and create the incoming DStream
    val configIn = new Properties
    configIn.put("infinispan.rdd.cacheName", inputDataGridCache)
    configIn.put("infinispan.client.hotrod.server_list", dataGridServer)
    val ispnStream = new InfinispanInputDStream[String, ShipmentData](ssc, StorageLevel.MEMORY_ONLY, configIn)

Next section is the kernel of this class, it extracts data from incoming stream and call the mapFunc function to calculate the average sensor data.

    // extract the (place, temperature) pair from the incoming stream that is composed of (key, value, eventType)
    // we are assuming only events of type CLIENT_CACHE_ENTRY_CREATED will be present.
    val measurementStream: DStream[(String, ShipmentData)] = ispnStream map { case (key, value, eventType) => (key, value) }

    // as measurements are batched, it can contain several measurements from the same place - let's group them together
    val measurementGrouped: DStream[(String, Iterable[ShipmentData])] = measurementStream.groupByKey()

    // Produce a new DStream combining the previous DStream with the stored state
    val avgSensorDataStream: MapWithStateDStream[String, Iterable[ShipmentData], Map[String, Iterable[ShipmentData]], (String, ShipmentData)] =
      measurementGrouped.mapWithState(StateSpec.function(mapFunc))

Last part of the main method will just write to the JDG 7 cache (avg-sensor-data).

    // write stream to the avg-sensor-data cache
    val configOut = new Properties
    configOut.put("infinispan.rdd.cacheName", outputDataGridCache)
    configOut.put("infinispan.client.hotrod.server_list", dataGridServer)
    avgSensorDataStream.writeToInfinispan(configOut)

The other funciton (mapFunc) in this class focuses on the getting the past 5-minute sensor data and calculating averages based on it.

First, get the existing statemap, and initialize other variables like the new average shipment data collection.

    // obtain the state map contains the past 5 minutes shipment data
    val stateMap: Map[String, Iterable[ShipmentData]] = state.getOption().getOrElse(Map[String, Iterable[ShipmentData]]())
    val pastData: (Iterable[ShipmentData]) = stateMap.getOrElse(place, (Iterable[ShipmentData])())

    var sumTemp: Double = 0d;
    var sumHumd: Double = 0d;
    val currTime: Long = java.lang.System.currentTimeMillis();
    var skuNos: String = "";

    //create the new list for storing the data, from both pastData and currData
    var newShipmentData = collection.mutable.ListBuffer[ShipmentData]();

Then get the total sum from both history data and current data.

    //only count the time in less than 5 minutes
    pastData.foreach { data: ShipmentData =>
      val sensorTime: Long = data.getSensorTime();

      //for testing usage
      val sensorTimeInDisplay = new Date(sensorTime);

      //only include the data from past 5 minutes
      if (currTime - sensorTime < 5 * 60 * 1000) {
        sumTemp = sumTemp + data.getTemperature();
        sumHumd = sumHumd + data.getHumidity();
        newShipmentData += data;
      }
    }

    val currData: Iterable[ShipmentData] = currDataLists.getOrElse(List());

    currData.foreach { data: ShipmentData =>
      //only count the time in less than 5 minutes
      val sensorTime: Long = data.getSensorTime();

      //for testing usage
      val sensorTimeInDisplay = new Date(sensorTime);
      //only include the data from past 5 minutes
      if (currTime - sensorTime < 5 * 60 * 1000) {
        sumTemp = sumTemp + data.getTemperature();
        sumHumd = sumHumd + data.getHumidity();
        newShipmentData += data;
        skuNos = data.getSkuNumbers();
      }
    }

Last part calculates the average data and updates statemap.

    val count = newShipmentData.size;
     var avgTemp: Double = sumTemp / count;
    var avgHumd: Double = sumHumd / count;

    var avgShipmentData: ShipmentData = new ShipmentData();
    avgShipmentData.setTemperature(avgTemp);
    avgShipmentData.setHumidity(avgHumd);
    avgShipmentData.setSkuNumbers(skuNos);
    avgShipmentData.setSensorTime(currTime);

    // update stored state
    state.update(stateMap.updated(place, newShipmentData.toList))

    (place, avgShipmentData)

The first part of Client’s main method is the java argument handling, which deals with the IP addresses and shipment ids that it needs to listen to.

      // check provided arguments - at least one shipment needs to be specified
      if (args.length < 1) {
         System.err.println("You have to provide list of shipment to watch, at least one!");
         System.exit(1);
      }
      String DATAGRID_IP = args[0];

It corrosponds to the invoke command of Client.java

# java -jar target/temperature-client-jar-with-dependencies.jar 10.19.137.34 shipment1 shipment5 shipment9

Those shipment id strings that the client is interested in, will be put into a collection, then passed to the second class, AvgSensorDataListener.

      Set<String> shipmentToWatch = new HashSet<>(args.length);
      Collections.addAll(shipmentToWatch, args);

The next part will create the Hot Rod Java Client to JDG cluster and connect with the sensor-avg-data JDG cache.

// Configure remote cache
      ConfigurationBuilder builder = new ConfigurationBuilder();
      builder.addServer().host(DATAGRID_IP).port(ConfigurationProperties.DEFAULT_HOTROD_PORT);
      RemoteCacheManager cacheManager = new RemoteCacheManager(builder.build());
      RemoteCache<String, ShipmentData> cache = cacheManager.getCache(CACHE_NAME);

Then it just registers the listener:

// Add cache listener and wait for specified amount of time
      AvgSensorDataListener avgListener = new AvgSensorDataListener(cache, shipmentToWatch);
      cache.addClientListener(avgListener);
      int LISTEN_TIME = 24; // how long the client should listen to changes, in hours
      System.out.printf("Client will be listening to avg. sensor updates for %d hours%n", LISTEN_TIME);
      Thread.sleep(LISTEN_TIME * 60 * 60 * 1000);

The second class in Client package is AvgSensorDataListener.java

@ClientListener
public class AvgSensorDataListener {
	private final RemoteCache<String, ShipmentData> cache;
	private final Set<String> watchedShipment;

	public AvgSensorDataListener(RemoteCache<String, ShipmentData> cache, Set<String> watchedShipment) {
		this.cache = cache;
		this.watchedShipment = watchedShipment;
	}

	@ClientCacheEntryCreated
	public void entryCreated(ClientCacheEntryCreatedEvent<String> event) {
		for (String oneWatchedShipment : watchedShipment){
			if (event.getKey()!=null && oneWatchedShipment != null && event.getKey().contains(oneWatchedShipment))
				judgeAction(event.getKey());
		}

	}

	@ClientCacheEntryModified
	public void entryModified(ClientCacheEntryModifiedEvent<String> event) {
		for (String oneWatchedShipment : watchedShipment){
			if (event.getKey()!=null && oneWatchedShipment != null && event.getKey().contains(oneWatchedShipment))
				judgeAction(event.getKey());
		}
	}

	private void judgeAction(String key) {
		ShipmentData data = cache.get(key);
		// setup a few test business rules, rules are based on SKU #.
		String skuNo = data.getSkuNumbers();
		if (skuNo.contains("meat") && data.getTemperature() > 15) {
			// business logic: meat's average temperature can't be higher than 15
			System.out.println("Average data is from: " + key + data.getValues());
			System.out.println(" Warning, MEAT shipment is in danger,  current temperature: " + (int) data.getTemperature() + " threshold is 15");
			System.out.println("");
		}
		if (skuNo.contains("metal") && data.getHumidity() > 30) {
			// business logic:metal's average humidity can't be higher than 30
			System.out.println("Average data is from: " + key + data.getValues());
			System.out.println("  Warning, METAL shipment is in danger,  current humidity: " + (int) data.getHumidity() + " threshold is 30");
			System.out.println("");
		}
		if (skuNo.contains("animal") && (data.getTemperature() < 15 || data.getTemperature() > 30 || data.getHumidity() < 20 || data.getHumidity()> 90)) {
			// business logic: animal average temperature can't be lower than 15 or higher than 30 and average humidity need to between 20 and 90
			System.out.println("Average data is from: " + key + data.getValues());
			System.out.println(" Warning, ANIMAL shipment is in danger, current temperature:  " + (int) data.getTemperature() + " threshold is between 15 to 30" + " current humidity: " + (int) data.getHumidity() + " threshold is between 20 to 90");
			System.out.println("");
		}

	}
}

Both entryCreated and entryModified methods will compare the watched shipment string with the current created/modified shipment string. If they are a match, then it invokes another method to see if the data will trigger the alert.

	@ClientCacheEntryCreated
	public void entryCreated(ClientCacheEntryCreatedEvent<String> event) {
		for (String oneWatchedShipment : watchedShipment){
			if (event.getKey()!=null && oneWatchedShipment != null && event.getKey().contains(oneWatchedShipment))
				updateAction(event.getKey());
		}

	}

	@ClientCacheEntryModified
	public void entryModified(ClientCacheEntryModifiedEvent<String> event) {
		for (String oneWatchedShipment : watchedShipment){
			if (event.getKey()!=null && oneWatchedShipment != null && event.getKey().contains(oneWatchedShipment))
				updateAction(event.getKey());
		}
	}

When the right sensor data is encountered, the judgeAction method will be invoked. It uses the SKU number to decide whether the threshold is reached or not, to then send out an alert. Below is the code for meat shipment; the other cases are similar.

		ShipmentData data = cache.get(key);
		// setup a few test business rules, rules are based on SKU #.
		String skuNo = data.getSkuNumbers();
		if (skuNo.contains("meat") && data.getTemperature() > 15) {
			// business logic: meat's average temperature can't be higher than 15
			System.out.println("Average data is from: " + key + data.getValues());
			System.out.println(" Warning, MEAT shipment is in danger,  current temperature: " + (int) data.getTemperature() + " threshold is 15");
			System.out.println("");
		}