MorphlineUtils { @SuppressWarnings("serial") public static FlatMapFunction<Row, Row> morphlineMapper(final String morphlineFile, final String morphlineId, final StructType outputSchema, final boolean errorOnEmpty) { return new FlatMapFunction<Row, Row>() { @Override public Iterator<Row> call(Row row) throws Exception { Pipeline pipeline = MorphlineUtils.getPipeline(morphlineFile, morphlineId); if (null == pipeline) { pipeline = MorphlineUtils.setPipeline(morphlineFile, morphlineId, new Collector(), true); } StructType inputSchema = row.schema(); if (null == inputSchema) { throw new RuntimeException("Row does not have an associated StructType schema"); } Record inputRecord = new Record(); String[] fieldNames = inputSchema.fieldNames(); for (int i = 0; i < fieldNames.length; i++) { inputRecord.put(fieldNames[i], row.get(i)); } List<Record> outputRecords = MorphlineUtils.executePipeline(pipeline, inputRecord, errorOnEmpty); List<Row> outputRows = Lists.newArrayListWithCapacity(outputRecords.size()); for (Record record : outputRecords) { outputRows.add(MorphlineUtils.convertToRow(outputSchema, record)); } return outputRows.iterator(); } }; }  static Pipeline getPipeline(String morphlineFile, String morphlineId); static Pipeline setPipeline(String morphlineFile, String morphlineId, Collector collector, boolean isProduction); static List<Record> executePipeline(Pipeline pipeline, Record inputRecord); static List<Record> executePipeline(Pipeline pipeline, Record inputRecord, boolean errorOnEmpty); @SuppressWarnings("serial") static FlatMapFunction<Row, Row> morphlineMapper(final String morphlineFile, final String morphlineId,
                                                          final StructType outputSchema, final boolean errorOnEmpty); static Row convertToRow(StructType schema, Record record);  }
@Test public void morphlineMapper( final @Mocked MorphlineUtils.Pipeline pipeline, final @Mocked Row row, final @Mocked StructType schema ) throws Exception { new Expectations(MorphlineUtils.class) {{ MorphlineUtils.getPipeline("file", "id"); result = pipeline; times = 1; MorphlineUtils.executePipeline(pipeline, (Record) any, true); result = Lists.newArrayList(); times = 1; row.schema(); result = schema; row.get(anyInt); returns("val1", "val2"); times = 2; schema.fieldNames(); result = new String[] { "one", "two"}; }}; FlatMapFunction<Row, Row> function = MorphlineUtils.morphlineMapper("file", "id", schema, true); Iterator<Row> results = function.call(row); assertEquals("Invalid number of Rows returned", 0, Lists.newArrayList(results).size()); new Verifications() {{ Record record; MorphlineUtils.executePipeline(pipeline, record = withCapture(), true); assertEquals(2, record.getFields().size()); assertEquals("val1", record.get("one").get(0)); }}; }