@Test public void testPipelineWithMoreSimpleTypes() throws Exception { String[] commands = new String[] { "parse-as-csv __col ,", "drop __col", "set columns first,last,email,timestamp,weight" }; Schema schema = Schema.recordOf( "output", Schema.Field.of("first", Schema.of(Schema.Type.STRING)), Schema.Field.of("last", Schema.of(Schema.Type.STRING)), Schema.Field.of("email", Schema.of(Schema.Type.STRING)), Schema.Field.of("timestamp", Schema.of(Schema.Type.LONG)), Schema.Field.of("weight", Schema.of(Schema.Type.FLOAT)) ); RecipePipeline pipeline = TestingRig.execute(commands); Row row = new Row("__col", new String("Larry,Perez,lperezqt@umn.edu,1481666448,186.66")); StructuredRecord record = (StructuredRecord) pipeline.execute(Arrays.asList(row), schema).get(0); Assert.assertEquals("Larry", record.get("first")); Assert.assertEquals("Perez", record.get("last")); Assert.assertEquals("lperezqt@umn.edu", record.get("email")); Assert.assertEquals(1481666448L, record.<Long>get("timestamp").longValue()); Assert.assertEquals(186.66f, record.get("weight"), 0.0001f); }
@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();  }