Application implements StreamingApplication { @Override public void populateDAG(DAG dag, Configuration conf) { locality = Locality.CONTAINER_LOCAL; dag.getAttributes().put(DAG.STREAMING_WINDOW_SIZE_MILLIS, 1000); SimpleSinglePortZeroMQPullStringInputOperator input = dag.addOperator("input", new SimpleSinglePortZeroMQPullStringInputOperator(addr)); ApacheLogParseOperator parse = dag.addOperator("parse", new ApacheLogParseOperator()); UniqueCounter<String> ipAddrCount = dag.addOperator("ipAddrCount", new UniqueCounter<String>()); UniqueCounter<String> urlCount = dag.addOperator("urlCount", new UniqueCounter<String>()); UniqueCounter<String> httpStatusCount = dag.addOperator("httpStatusCount", new UniqueCounter<String>()); Sum<Long> numOfBytesSum = dag.addOperator("numOfBytesSum", new Sum<Long>()); dag.getMeta(numOfBytesSum).getAttributes().put(OperatorContext.APPLICATION_WINDOW_COUNT, 3); dag.addStream("input-parse", input.outputPort, parse.data).setLocality(locality); dag.addStream("parse-ipAddrCount", parse.outputIPAddress, ipAddrCount.data).setLocality(locality); dag.addStream("parse-urlCount", parse.outputUrl, urlCount.data).setLocality(locality); dag.addStream("parse-httpStatusCount", parse.outputStatusCode, httpStatusCount.data).setLocality(locality); dag.addStream("parse-numOfBytesSum", parse.outputBytes, numOfBytesSum.data).setLocality(locality); ConsoleOutputOperator consoleOperator1 = dag.addOperator("console1", new ConsoleOutputOperator()); ConsoleOutputOperator consoleOperator2 = dag.addOperator("console2", new ConsoleOutputOperator()); ConsoleOutputOperator consoleOperator3 = dag.addOperator("console3", new ConsoleOutputOperator()); ConsoleOutputOperator consoleOperator4 = dag.addOperator("console4", new ConsoleOutputOperator()); dag.addStream("ipAddrCount-console", ipAddrCount.count, consoleOperator1.input); dag.addStream("urlCount-console", urlCount.count, consoleOperator2.input); dag.addStream("httpStatusCount-console", httpStatusCount.count, consoleOperator3.input); dag.addStream("numOfBytesSum-console", numOfBytesSum.sumLong, consoleOperator4.input); }  @Override void populateDAG(DAG dag, Configuration conf);  }
@Test public void testSomeMethod() throws Exception { Application app = new Application(); LocalMode lma = LocalMode.newInstance(); app.populateDAG(lma.getDAG(), new Configuration(false)); final LocalMode.Controller lc = lma.getController(); long start = System.currentTimeMillis(); lc.run(); long end = System.currentTimeMillis(); long time = end - start; LOG.info("Test used " + time + " ms"); }