@Test public void testDoubleInput() throws Exception { Stitch func = new Stitch(); DataBag b1 = BagFactory.getInstance().newDefaultBag(); Tuple t = TupleFactory.getInstance().newTuple(); t.append("a"); t.append("b"); b1.add(t); t = TupleFactory.getInstance().newTuple(); t.append("c"); t.append("d"); b1.add(t); DataBag b2 = BagFactory.getInstance().newDefaultBag(); t = TupleFactory.getInstance().newTuple(); t.append("1"); t.append("2"); b2.add(t); t = TupleFactory.getInstance().newTuple(); t.append("3"); t.append("4"); b2.add(t); t = TupleFactory.getInstance().newTuple(); t.append(b1); t.append(b2); DataBag out = func.exec(t); assertEquals(2, out.size()); Iterator<Tuple> iter = out.iterator(); t = iter.next(); assertEquals(4, t.size()); assertEquals("a", t.get(0)); assertEquals("b", t.get(1)); assertEquals("1", t.get(2)); assertEquals("2", t.get(3)); t = iter.next(); assertEquals(4, t.size()); assertEquals("c", t.get(0)); assertEquals("d", t.get(1)); assertEquals("3", t.get(2)); assertEquals("4", t.get(3)); }
@Override public DataBag exec(Tuple input) throws IOException { if (input == null || input.size() == 0) return null; List<DataBag> bags = new ArrayList<DataBag>(input.size()); for (int i = 0; i < input.size(); i++) { Object o = input.get(i); try { bags.add((DataBag)o); } catch (ClassCastException cce) { int errCode = 2107; String msg = "Stitch expected bags as input but argument " + i + " is a " + DataType.findTypeName(o); throw new ExecException(msg, errCode, PigException.INPUT); } } if (bags.size() == 1) return bags.get(0); DataBag output = BagFactory.getInstance().newDefaultBag(); List<Iterator<Tuple>> iters = new ArrayList<Iterator<Tuple>>(bags.size()); for (DataBag bag : bags) { iters.add(bag.iterator()); } while (iters.get(0).hasNext()) { Tuple outTuple = TupleFactory.getInstance().newTuple(); for (Iterator<Tuple> iter : iters) { if (iter.hasNext()) { Tuple t = iter.next(); List<Object> fields = t.getAll(); for (Object field : fields) { outTuple.append(field); } } } output.add(outTuple); } return output; }
Stitch extends EvalFunc<DataBag> { @Override public DataBag exec(Tuple input) throws IOException { if (input == null || input.size() == 0) return null; List<DataBag> bags = new ArrayList<DataBag>(input.size()); for (int i = 0; i < input.size(); i++) { Object o = input.get(i); try { bags.add((DataBag)o); } catch (ClassCastException cce) { int errCode = 2107; String msg = "Stitch expected bags as input but argument " + i + " is a " + DataType.findTypeName(o); throw new ExecException(msg, errCode, PigException.INPUT); } } if (bags.size() == 1) return bags.get(0); DataBag output = BagFactory.getInstance().newDefaultBag(); List<Iterator<Tuple>> iters = new ArrayList<Iterator<Tuple>>(bags.size()); for (DataBag bag : bags) { iters.add(bag.iterator()); } while (iters.get(0).hasNext()) { Tuple outTuple = TupleFactory.getInstance().newTuple(); for (Iterator<Tuple> iter : iters) { if (iter.hasNext()) { Tuple t = iter.next(); List<Object> fields = t.getAll(); for (Object field : fields) { outTuple.append(field); } } } output.add(outTuple); } return output; } }
Stitch extends EvalFunc<DataBag> { @Override public DataBag exec(Tuple input) throws IOException { if (input == null || input.size() == 0) return null; List<DataBag> bags = new ArrayList<DataBag>(input.size()); for (int i = 0; i < input.size(); i++) { Object o = input.get(i); try { bags.add((DataBag)o); } catch (ClassCastException cce) { int errCode = 2107; String msg = "Stitch expected bags as input but argument " + i + " is a " + DataType.findTypeName(o); throw new ExecException(msg, errCode, PigException.INPUT); } } if (bags.size() == 1) return bags.get(0); DataBag output = BagFactory.getInstance().newDefaultBag(); List<Iterator<Tuple>> iters = new ArrayList<Iterator<Tuple>>(bags.size()); for (DataBag bag : bags) { iters.add(bag.iterator()); } while (iters.get(0).hasNext()) { Tuple outTuple = TupleFactory.getInstance().newTuple(); for (Iterator<Tuple> iter : iters) { if (iter.hasNext()) { Tuple t = iter.next(); List<Object> fields = t.getAll(); for (Object field : fields) { outTuple.append(field); } } } output.add(outTuple); } return output; }  }
Stitch extends EvalFunc<DataBag> { @Override public DataBag exec(Tuple input) throws IOException { if (input == null || input.size() == 0) return null; List<DataBag> bags = new ArrayList<DataBag>(input.size()); for (int i = 0; i < input.size(); i++) { Object o = input.get(i); try { bags.add((DataBag)o); } catch (ClassCastException cce) { int errCode = 2107; String msg = "Stitch expected bags as input but argument " + i + " is a " + DataType.findTypeName(o); throw new ExecException(msg, errCode, PigException.INPUT); } } if (bags.size() == 1) return bags.get(0); DataBag output = BagFactory.getInstance().newDefaultBag(); List<Iterator<Tuple>> iters = new ArrayList<Iterator<Tuple>>(bags.size()); for (DataBag bag : bags) { iters.add(bag.iterator()); } while (iters.get(0).hasNext()) { Tuple outTuple = TupleFactory.getInstance().newTuple(); for (Iterator<Tuple> iter : iters) { if (iter.hasNext()) { Tuple t = iter.next(); List<Object> fields = t.getAll(); for (Object field : fields) { outTuple.append(field); } } } output.add(outTuple); } return output; }  @Override DataBag exec(Tuple input); @Override Schema outputSchema(Schema inputSch); }
Stitch extends EvalFunc<DataBag> { @Override public DataBag exec(Tuple input) throws IOException { if (input == null || input.size() == 0) return null; List<DataBag> bags = new ArrayList<DataBag>(input.size()); for (int i = 0; i < input.size(); i++) { Object o = input.get(i); try { bags.add((DataBag)o); } catch (ClassCastException cce) { int errCode = 2107; String msg = "Stitch expected bags as input but argument " + i + " is a " + DataType.findTypeName(o); throw new ExecException(msg, errCode, PigException.INPUT); } } if (bags.size() == 1) return bags.get(0); DataBag output = BagFactory.getInstance().newDefaultBag(); List<Iterator<Tuple>> iters = new ArrayList<Iterator<Tuple>>(bags.size()); for (DataBag bag : bags) { iters.add(bag.iterator()); } while (iters.get(0).hasNext()) { Tuple outTuple = TupleFactory.getInstance().newTuple(); for (Iterator<Tuple> iter : iters) { if (iter.hasNext()) { Tuple t = iter.next(); List<Object> fields = t.getAll(); for (Object field : fields) { outTuple.append(field); } } } output.add(outTuple); } return output; }  @Override DataBag exec(Tuple input); @Override Schema outputSchema(Schema inputSch);  }