Updating the Prediction Model

punctuateMethodForProcessor.java

// process call left out for clarity; it places airline data
// in a list, storing in state store by airport code
@Override
public void punctuate(long timestamp) {
  KeyValueIterator<String, List<String>> allFlights = flights.all();
	while (allFlights.hasNext()) {
  	KeyValue<String, List<String>> kv = allFlights.next();
  	List<String> flightList = kv.value;
  	String key = kv.key;
	if(flightList.size() >= 100) {
    	try {
          byte[] serializedRegression = ModelBuilder.train(flightList);
          context().forward(key, serializedRegression);
          LOG.info("updating model for {}", key);
          flightList.clear();
          flights.put(key, flightList);
   } catch (Exception e) {
    LOG.error("couldn't update online regression for {}",key, e);
    }
	}
  }
  allFlights.close();
}