ApplicationWithDCWithoutDeserializer extends ApplicationDimensionComputation { @Override public void populateDAG(DAG dag, Configuration configuration) { redisServer = configuration.get("dt.application.AppWithDCWithoutDe.redisServer"); DefaultOutputPort<DimensionTuple> upstreamOutput = populateUpstreamDAG(dag, configuration); populateDimensionsDAG(dag, configuration, upstreamOutput); } ApplicationWithDCWithoutDeserializer(); @Override void populateDAG(DAG dag, Configuration configuration); DefaultOutputPort<DimensionTuple> populateUpstreamDAG(DAG dag, Configuration configuration); String getRedisServer(); void setRedisServer(String redisServer); static final String APP_NAME; }
@Test public void test() throws Exception { Configuration conf = new Configuration(false); conf.set(PROP_STORE_PATH, "target/tmp"); LocalMode lma = LocalMode.newInstance(); DAG dag = lma.getDAG(); super.populateDAG(dag, conf); StreamingApplication app = new StreamingApplication() { @Override public void populateDAG(DAG dag, Configuration conf) { } }; lma.prepareDAG(app, conf); final LocalMode.Controller lc = lma.getController(); lc.run(600000); lc.shutdown(); }