StringBagsMerger extends EvalFunc<DataBag> { @Override public DataBag exec(Tuple tuple) throws IOException { if (tuple == null || tuple.size() == 0) { return null; } List<Tuple> tuples = new ArrayList<Tuple>(); for (int i = 0; i < tuple.size(); i++) { if (tuple.get(i) != null) { DataBag db = (DataBag) tuple.get(i); Iterator<Tuple> it = db.iterator(); while (it.hasNext()) { Tuple next = it.next(); if (!tuples.contains(next)) { tuples.add(next); } } } } if (tuples.isEmpty()) { return null; } BagFactory bagFactory = BagFactory.getInstance(); return bagFactory.newDefaultBag(Lists.newArrayList(tuples)); }  @Override DataBag exec(Tuple tuple); @Override Schema outputSchema(Schema input);  }
@Test public void testUDF() throws IOException { StringBagsMerger udf = new StringBagsMerger(); TupleFactory tupleFactory = TupleFactory.getInstance(); BagFactory bagFactory = BagFactory.getInstance(); DataBag emptyBag = bagFactory.newDefaultBag(); DataBag bag1 = bagFactory.newDefaultBag(Lists.newArrayList(tupleFactory.newTuple("tup1"), tupleFactory.newTuple("tup2"))); DataBag bag2 = bagFactory.newDefaultBag(Lists.newArrayList(tupleFactory.newTuple("tup3"), tupleFactory.newTuple("tup4"))); DataBag bag3 = bagFactory.newDefaultBag(Lists.newArrayList(tupleFactory.newTuple("tup1"), tupleFactory.newTuple("tup4"), tupleFactory.newTuple("tup5"))); DataBag bag4 = bagFactory.newDefaultBag(Lists.newArrayList(tupleFactory.newTuple("tup1"), tupleFactory.newTuple("tup2"), tupleFactory.newTuple("tup3"), tupleFactory.newTuple("tup4"), tupleFactory.newTuple("tup5"))); assertNull(udf.exec(null)); assertNull(udf.exec(tupleFactory.newTuple())); assertNull(udf.exec(tupleFactory.newTuple((DataBag)null))); assertNull(udf.exec(tupleFactory.newTuple(emptyBag))); assertEquals(bag1, udf.exec(tupleFactory.newTuple(bag1))); assertEquals(bag4, udf.exec(tupleFactory.newTuple(Lists.newArrayList(bag1, emptyBag, bag2, bag3)))); }