Skip to content

Instantly share code, notes, and snippets.

Embed
What would you like to do?
Updating the Prediction Model
// 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();
}
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment
You can’t perform that action at this time.