@Test public void testSchema() throws Exception { Schema s = new Schema(); Schema in = new Schema(); s.add(new FieldSchema("x", DataType.CHARARRAY)); s.add(new FieldSchema("y", DataType.INTEGER)); in.add(new FieldSchema("A", s, DataType.BAG)); FieldSchema fs = new FieldSchema("Over", new Schema(Schema.generateNestedSchema(DataType.BAG, DataType.NULL)), DataType.BAG); in.add(fs); Stitch func = new Stitch(); Schema out = func.outputSchema(in); assertEquals("{stitched: {x: chararray,y: int,{NULL}}}", out.toString()); }
@Override public Schema outputSchema(Schema inputSch) { List<FieldSchema> fields = inputSch.getFields(); try { Schema bagSchema = null; for (FieldSchema field : fields) { if (field.type != DataType.BAG) { throw new RuntimeException( "Only bags should be passed to " + "Stitch, schema indicates a " + DataType.findTypeName(field.type) + " being passed"); } if (bagSchema == null && field.schema != null ) { if (field.schema.getField(0).type == DataType.TUPLE) { field = field.schema.getField(0); } bagSchema = new Schema(field.schema); } else if (field.schema != null) { if (field.schema.getField(0).type == DataType.TUPLE) { field = field.schema.getField(0); } for (FieldSchema fs : field.schema.getFields()) { bagSchema.add(fs); } } } FieldSchema outFS = new FieldSchema("stitched", bagSchema, DataType.BAG); return new Schema(outFS); } catch (FrontendException fe) { throw new RuntimeException("Unable to create nested schema", fe); } }
Stitch extends EvalFunc<DataBag> { @Override public Schema outputSchema(Schema inputSch) { List<FieldSchema> fields = inputSch.getFields(); try { Schema bagSchema = null; for (FieldSchema field : fields) { if (field.type != DataType.BAG) { throw new RuntimeException( "Only bags should be passed to " + "Stitch, schema indicates a " + DataType.findTypeName(field.type) + " being passed"); } if (bagSchema == null && field.schema != null ) { if (field.schema.getField(0).type == DataType.TUPLE) { field = field.schema.getField(0); } bagSchema = new Schema(field.schema); } else if (field.schema != null) { if (field.schema.getField(0).type == DataType.TUPLE) { field = field.schema.getField(0); } for (FieldSchema fs : field.schema.getFields()) { bagSchema.add(fs); } } } FieldSchema outFS = new FieldSchema("stitched", bagSchema, DataType.BAG); return new Schema(outFS); } catch (FrontendException fe) { throw new RuntimeException("Unable to create nested schema", fe); } } }
Stitch extends EvalFunc<DataBag> { @Override public Schema outputSchema(Schema inputSch) { List<FieldSchema> fields = inputSch.getFields(); try { Schema bagSchema = null; for (FieldSchema field : fields) { if (field.type != DataType.BAG) { throw new RuntimeException( "Only bags should be passed to " + "Stitch, schema indicates a " + DataType.findTypeName(field.type) + " being passed"); } if (bagSchema == null && field.schema != null ) { if (field.schema.getField(0).type == DataType.TUPLE) { field = field.schema.getField(0); } bagSchema = new Schema(field.schema); } else if (field.schema != null) { if (field.schema.getField(0).type == DataType.TUPLE) { field = field.schema.getField(0); } for (FieldSchema fs : field.schema.getFields()) { bagSchema.add(fs); } } } FieldSchema outFS = new FieldSchema("stitched", bagSchema, DataType.BAG); return new Schema(outFS); } catch (FrontendException fe) { throw new RuntimeException("Unable to create nested schema", fe); } }  }
Stitch extends EvalFunc<DataBag> { @Override public Schema outputSchema(Schema inputSch) { List<FieldSchema> fields = inputSch.getFields(); try { Schema bagSchema = null; for (FieldSchema field : fields) { if (field.type != DataType.BAG) { throw new RuntimeException( "Only bags should be passed to " + "Stitch, schema indicates a " + DataType.findTypeName(field.type) + " being passed"); } if (bagSchema == null && field.schema != null ) { if (field.schema.getField(0).type == DataType.TUPLE) { field = field.schema.getField(0); } bagSchema = new Schema(field.schema); } else if (field.schema != null) { if (field.schema.getField(0).type == DataType.TUPLE) { field = field.schema.getField(0); } for (FieldSchema fs : field.schema.getFields()) { bagSchema.add(fs); } } } FieldSchema outFS = new FieldSchema("stitched", bagSchema, DataType.BAG); return new Schema(outFS); } catch (FrontendException fe) { throw new RuntimeException("Unable to create nested schema", fe); } }  @Override DataBag exec(Tuple input); @Override Schema outputSchema(Schema inputSch); }
Stitch extends EvalFunc<DataBag> { @Override public Schema outputSchema(Schema inputSch) { List<FieldSchema> fields = inputSch.getFields(); try { Schema bagSchema = null; for (FieldSchema field : fields) { if (field.type != DataType.BAG) { throw new RuntimeException( "Only bags should be passed to " + "Stitch, schema indicates a " + DataType.findTypeName(field.type) + " being passed"); } if (bagSchema == null && field.schema != null ) { if (field.schema.getField(0).type == DataType.TUPLE) { field = field.schema.getField(0); } bagSchema = new Schema(field.schema); } else if (field.schema != null) { if (field.schema.getField(0).type == DataType.TUPLE) { field = field.schema.getField(0); } for (FieldSchema fs : field.schema.getFields()) { bagSchema.add(fs); } } } FieldSchema outFS = new FieldSchema("stitched", bagSchema, DataType.BAG); return new Schema(outFS); } catch (FrontendException fe) { throw new RuntimeException("Unable to create nested schema", fe); } }  @Override DataBag exec(Tuple input); @Override Schema outputSchema(Schema inputSch);  }