@Test public void testPipeline() throws Exception { String[] commands = new String[] { "parse-as-csv __col ,", "drop __col", "set columns a,b,c,d,e,f,g", "rename a first", "drop b" }; Schema schema = Schema.recordOf( "output", Schema.Field.of("first", Schema.of(Schema.Type.STRING)), Schema.Field.of("c", Schema.of(Schema.Type.STRING)), Schema.Field.of("d", Schema.of(Schema.Type.STRING)), Schema.Field.of("e", Schema.of(Schema.Type.STRING)), Schema.Field.of("f", Schema.of(Schema.Type.STRING)), Schema.Field.of("g", Schema.of(Schema.Type.STRING)) ); RecipePipeline pipeline = TestingRig.execute(commands); Row row = new Row("__col", new String("a,b,c,d,e,f,1.0")); StructuredRecord record = (StructuredRecord) pipeline.execute(Arrays.asList(row), schema).get(0); Assert.assertEquals("a", record.get("first")); Assert.assertEquals("c", record.get("c")); Assert.assertEquals("d", record.get("d")); Assert.assertEquals("e", record.get("e")); Assert.assertEquals("f", record.get("f")); Assert.assertEquals("1.0", record.get("g")); }
@Override public List<StructuredRecord> execute(List<Row> rows, Schema schema) throws RecipeException { rows = execute(rows); try { List<StructuredRecord> output = convertor.toStructureRecord(rows, schema); return output; } catch (RecordConvertorException e) { throw new RecipeException("Problem converting into output record. Reason : " + e.getMessage(), e); } }
RecipePipelineExecutor implements RecipePipeline<Row, StructuredRecord, ErrorRecord> { @Override public List<StructuredRecord> execute(List<Row> rows, Schema schema) throws RecipeException { rows = execute(rows); try { List<StructuredRecord> output = convertor.toStructureRecord(rows, schema); return output; } catch (RecordConvertorException e) { throw new RecipeException("Problem converting into output record. Reason : " + e.getMessage(), e); } } }
RecipePipelineExecutor implements RecipePipeline<Row, StructuredRecord, ErrorRecord> { @Override public List<StructuredRecord> execute(List<Row> rows, Schema schema) throws RecipeException { rows = execute(rows); try { List<StructuredRecord> output = convertor.toStructureRecord(rows, schema); return output; } catch (RecordConvertorException e) { throw new RecipeException("Problem converting into output record. Reason : " + e.getMessage(), e); } }  }
RecipePipelineExecutor implements RecipePipeline<Row, StructuredRecord, ErrorRecord> { @Override public List<StructuredRecord> execute(List<Row> rows, Schema schema) throws RecipeException { rows = execute(rows); try { List<StructuredRecord> output = convertor.toStructureRecord(rows, schema); return output; } catch (RecordConvertorException e) { throw new RecipeException("Problem converting into output record. Reason : " + e.getMessage(), e); } }  @Override void initialize(RecipeParser parser, ExecutorContext context); @Override void destroy(); @Override List<StructuredRecord> execute(List<Row> rows, Schema schema); @Override List<Row> execute(List<Row> rows); @Override List<ErrorRecord> errors(); }
RecipePipelineExecutor implements RecipePipeline<Row, StructuredRecord, ErrorRecord> { @Override public List<StructuredRecord> execute(List<Row> rows, Schema schema) throws RecipeException { rows = execute(rows); try { List<StructuredRecord> output = convertor.toStructureRecord(rows, schema); return output; } catch (RecordConvertorException e) { throw new RecipeException("Problem converting into output record. Reason : " + e.getMessage(), e); } }  @Override void initialize(RecipeParser parser, ExecutorContext context); @Override void destroy(); @Override List<StructuredRecord> execute(List<Row> rows, Schema schema); @Override List<Row> execute(List<Row> rows); @Override List<ErrorRecord> errors();  }