ClassifyQuery extends BaseQueryFunction<MapState<Classifier<L>>, L> { @SuppressWarnings("unchecked") @Override public List<L> batchRetrieve(MapState<Classifier<L>> state, List<TridentTuple> tuples) { List<L> labels = new ArrayList<L>(); List<Classifier<L>> classifiers = state.multiGet(KeysUtil.toKeys(this.classifierName)); if (classifiers != null && !classifiers.isEmpty()) { Classifier<L> classifier = classifiers.get(0); if (classifier == null) { for (int i = 0; i < tuples.size(); i++) { labels.add(null); } } else { L label; Instance<L> instance; for (TridentTuple tuple : tuples) { instance = (Instance<L>) tuple.get(0); label = classifier.classify(instance.features); labels.add(label); } } } else { for (int i = 0; i < tuples.size(); i++) { labels.add(null); } } return labels; } ClassifyQuery(String classifierName); @SuppressWarnings("unchecked") @Override List<L> batchRetrieve(MapState<Classifier<L>> state, List<TridentTuple> tuples); void execute(TridentTuple tuple, L result, TridentCollector collector);  }
@Test public void should_classify_instance_using_classifier() { Boolean expectedLabel1 = true; Boolean expectedLabel2 = false; String classifierName = "TestLearner"; ClassifyQuery<Boolean> query = new ClassifyQuery<Boolean>(classifierName); 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); Classifier<Boolean> expectedClassifier = mock(Classifier.class); given(expectedClassifier.classify(same(features1))).willReturn(expectedLabel1); given(expectedClassifier.classify(same(features2))).willReturn(expectedLabel2); List<List<Object>> expectedKeys = asList(asList((Object) classifierName)); MapState<Classifier<Boolean>> state = mock(MapState.class); given(state.multiGet(expectedKeys)).willReturn(Arrays.asList(expectedClassifier)); List<Boolean> labels = query.batchRetrieve(state, tuples); assertEquals(2, labels.size()); assertTrue(labels.get(0)); assertFalse(labels.get(1)); }
@Test public void should_classify_instance_without_classifier() { String classifierName = "TestLearner"; ClassifyQuery<Boolean> query = new ClassifyQuery<Boolean>(classifierName); 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); MapState<Classifier<Boolean>> state = mock(MapState.class); given(state.multiGet(asList(asList((Object) classifierName)))).willReturn(EMPTY_LIST); List<Boolean> labels = query.batchRetrieve(state, tuples); assertEquals(2, labels.size()); assertNull(labels.get(0)); assertNull(labels.get(1)); }