MorphlineUtils { public static Pipeline setPipeline(String morphlineFile, String morphlineId, Collector collector, boolean isProduction) { LOG.debug("Constructing Pipeline[{}#{}]", morphlineFile, morphlineId); MorphlineContext context = new MorphlineContext.Builder() .setExceptionHandler(new FaultTolerance(isProduction, false)) .build(); Command morphline; try { morphline = new Compiler().compile( new File(morphlineFile), morphlineId, context, collector); } catch (Exception e) { throw new MorphlineCompilationException("Morphline compilation error", null, e); } Pipeline pipeline = new Pipeline(morphline, collector); JVMUtils.closeAtShutdown(pipeline); Notifications.notifyBeginTransaction(pipeline.getMorphline()); if (null == pipelineCache.get()) { pipelineCache.set(new HashMap<String, Pipeline>()); } pipelineCache.get().put(morphlineFile + SEPARATOR + morphlineId, pipeline); LOG.trace("Pipeline[{}#{}] prepared", morphlineFile, morphlineId); return pipeline; }  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 setPipeline() throws Exception { MorphlineUtils.setPipeline(getResourcePath(MORPHLINE_FILE), "default", new MorphlineUtils.Collector(), true); }
@Test (expected = MorphlineCompilationException.class) public void setPipelineFileNotFound() throws Exception { MorphlineUtils.setPipeline("file", "id", new MorphlineUtils.Collector(), true); }