RegressionQuery extends BaseQueryFunction<MapState<Regressor>, Double> { @SuppressWarnings("unchecked") @Override public List<Double> batchRetrieve(MapState<Regressor> state, List<TridentTuple> tuples) { List<Double> labels = new ArrayList<Double>(); List<Regressor> regressors = state.multiGet(KeysUtil.toKeys(this.regressorName)); if (regressors != null && !regressors.isEmpty()) { Regressor regressor = regressors.get(0); if (regressor == null) { for (int i = 0; i < tuples.size(); i++) { labels.add(null); } } else { Double label; Instance<Double> instance; for (TridentTuple tuple : tuples) { instance = (Instance<Double>) tuple.get(0); label = regressor.predict(instance.features); labels.add(label); } } } else { for (int i = 0; i < tuples.size(); i++) { labels.add(null); } } return labels; } RegressionQuery(String regressorName); @SuppressWarnings("unchecked") @Override List<Double> batchRetrieve(MapState<Regressor> state, List<TridentTuple> tuples); void execute(TridentTuple tuple, Double result, TridentCollector collector);  }
@Test public void should_classify_instance_using_classifier() { Double expectedPrediction1 = 2.5; Double expectedPrediction2 = 12.8; String regressorName = "TestLearner"; RegressionQuery query = new RegressionQuery(regressorName); double[] features1 = new double[10]; double[] features2 = new double[10]; TridentTuple tuple1 = createMockedInstanceTuple(features1); TridentTuple tuple2 = createMockedInstanceTuple(features2); List<TridentTuple> tuples = Arrays.asList(tuple1, tuple2); Regressor expectedRegressor = mock(Regressor.class); given(expectedRegressor.predict(same(features1))).willReturn(expectedPrediction1); given(expectedRegressor.predict(same(features2))).willReturn(expectedPrediction2); List<List<Object>> expectedKeys = asList(asList((Object) regressorName)); MapState<Regressor> state = mock(MapState.class); given(state.multiGet(expectedKeys)).willReturn(Arrays.asList(expectedRegressor)); List<Double> actualPredictions = query.batchRetrieve(state, tuples); assertEquals(2, actualPredictions.size()); assertEquals(expectedPrediction1, actualPredictions.get(0)); assertEquals(expectedPrediction2, actualPredictions.get(1)); }
@Test public void should_classify_instance_without_classifier() { String regressorName = "TestLearner"; RegressionQuery query = new RegressionQuery(regressorName); TridentTuple tuple1 = mock(TridentTuple.class); TridentTuple tuple2 = mock(TridentTuple.class); List<TridentTuple> tuples = Arrays.asList(tuple1, tuple2); List<List<Object>> expectedKeys = asList(asList((Object) regressorName)); MapState<Regressor> state = mock(MapState.class); given(state.multiGet(expectedKeys)).willReturn(EMPTY_LIST); List<Double> actualPredictions = query.batchRetrieve(state, tuples); assertEquals(2, actualPredictions.size()); assertNull(actualPredictions.get(0)); assertNull(actualPredictions.get(1)); }