InvertIndex extends BaseKeyValueOperator<K, V> implements Unifier<HashMap<V, ArrayList<K>>> { @Override public void endWindow() { for (Map.Entry<V, ArrayList<K>> e: map.entrySet()) { HashMap<V, ArrayList<K>> tuple = new HashMap<V, ArrayList<K>>(1); tuple.put(e.getKey(), e.getValue()); index.emit(tuple); } map.clear(); }  @Override void endWindow(); @Override void process(HashMap<V, ArrayList<K>> tuple); final transient DefaultInputPort<HashMap<K, V>> data; final transient DefaultOutputPort<HashMap<V, ArrayList<K>>> index; }
@SuppressWarnings({ "rawtypes", "unchecked" }) @Test public void testNodeProcessing() throws Exception { InvertIndex<String,String> oper = new InvertIndex<String,String>(); CollectorTestSink indexSink = new CollectorTestSink(); Sink inSink = oper.data.getSink(); oper.index.setSink(indexSink); oper.beginWindow(0); HashMap<String, String> input = new HashMap<String, String>(); input.put("a", "str"); input.put("b", "str"); inSink.put(input); input.clear(); input.put("a", "str1"); input.put("b", "str1"); inSink.put(input); input.clear(); input.put("c", "blah"); inSink.put(input); input.clear(); input.put("c", "str1"); inSink.put(input); oper.endWindow(); Assert.assertEquals("number emitted tuples", 3, indexSink.collectedTuples.size()); for (Object o: indexSink.collectedTuples) { log.debug(o.toString()); HashMap<String, ArrayList<String>> output = (HashMap<String, ArrayList<String>>)o; for (Map.Entry<String, ArrayList<String>> e: output.entrySet()) { String key = e.getKey(); ArrayList<String> alist = e.getValue(); if (key.equals("str1")) { Assert.assertEquals("Index for \"str1\" contains \"a\"", true, alist.contains("a")); Assert.assertEquals("Index for \"str1\" contains \"b\"", true, alist.contains("b")); Assert.assertEquals("Index for \"str1\" contains \"c\"", true, alist.contains("c")); } else if (key.equals("str")) { Assert.assertEquals("Index for \"str1\" contains \"a\"", true, alist.contains("a")); Assert.assertEquals("Index for \"str1\" contains \"b\"", true, alist.contains("b")); Assert.assertEquals("Index for \"str1\" contains \"c\"", false, alist.contains("c")); } else if (key.equals("blah")) { Assert.assertEquals("Index for \"str1\" contains \"a\"", false, alist.contains("a")); Assert.assertEquals("Index for \"str1\" contains \"b\"", false, alist.contains("b")); Assert.assertEquals("Index for \"str1\" contains \"c\"", true, alist.contains("c")); } } } }