IdConfidenceTupleDeduplicator extends EvalFunc<DataBag> { @Override public DataBag exec(Tuple tuple) throws IOException { if (tuple == null || tuple.size() == 0) { return null; } DataBag inputTuples = (DataBag) tuple.get(0); if (inputTuples != null && inputTuples.size() > 1) { Map<String, Tuple> deduplicatedTuplesMap = new TreeMap<String, Tuple>(); Iterator<Tuple> inputTuplesIterator = inputTuples.iterator(); while (inputTuplesIterator.hasNext()) { Tuple currentTuple = inputTuplesIterator.next(); updateStoredTupleWhenConfidenceLevelHigher(currentTuple, deduplicatedTuplesMap); } return BagFactory.getInstance().newDefaultBag(new ArrayList<Tuple>(deduplicatedTuplesMap.values())); } else { return inputTuples; } } IdConfidenceTupleDeduplicator();  IdConfidenceTupleDeduplicator(int confidenceLevelPosition);  IdConfidenceTupleDeduplicator(String confidenceLevelPosition); @Override DataBag exec(Tuple tuple); @Override Schema outputSchema(Schema input);  }
@Test public void testUDF() throws IOException { IdConfidenceTupleDeduplicator udf = new IdConfidenceTupleDeduplicator(); TupleFactory tupleFactory = TupleFactory.getInstance(); BagFactory bagFactory = BagFactory.getInstance(); DataBag emptyBag = bagFactory.newDefaultBag(); assertNull(udf.exec(null)); assertNull(udf.exec(tupleFactory.newTuple())); assertNull(udf.exec(tupleFactory.newTuple((DataBag)null))); assertEquals(emptyBag, udf.exec(tupleFactory.newTuple(emptyBag))); assertEquals( bagFactory.newDefaultBag(Collections.singletonList( tupleFactory.newTuple(Arrays.asList("tuple1", null)))), udf.exec(tupleFactory.newTuple( bagFactory.newDefaultBag(Collections.singletonList( tupleFactory.newTuple(Arrays.asList("tuple1", null))))))); assertEquals( bagFactory.newDefaultBag(Collections.singletonList( tupleFactory.newTuple(Arrays.asList("tuple1",0.9f)))), udf.exec(tupleFactory.newTuple( bagFactory.newDefaultBag(Lists.newArrayList( tupleFactory.newTuple(Arrays.asList("tuple1",0.9f)), tupleFactory.newTuple(Arrays.asList("tuple1",0.1f)), tupleFactory.newTuple(Arrays.asList("tuple1",null))))))); assertEquals( bagFactory.newDefaultBag(Lists.newArrayList( tupleFactory.newTuple(Arrays.asList("tuple1",null)), tupleFactory.newTuple(Arrays.asList("tuple2",0.1f)), tupleFactory.newTuple(Arrays.asList("tuple3",0.6f)))), udf.exec(tupleFactory.newTuple( bagFactory.newDefaultBag(Lists.newArrayList( tupleFactory.newTuple(Arrays.asList("tuple3",0.3f)), tupleFactory.newTuple(Arrays.asList("tuple2",0.1f)), tupleFactory.newTuple(Arrays.asList("tuple3",0.6f)), tupleFactory.newTuple(Arrays.asList("tuple1",null))))))); }