ApplicationLogGenerator implements StreamingApplication { @Override public void populateDAG(DAG dag, Configuration conf) { ApacheLogInputGenerator log = dag.addOperator("log", new ApacheLogInputGenerator()); log.setIpAddressFile("/com/datatorrent/contrib/apachelog/ipaddress.txt"); log.setUrlFile("/com/datatorrent/contrib/apachelog/urls.txt"); log.setAgentFile("/com/datatorrent/contrib/apachelog/agents.txt"); log.setRefererFile("/com/datatorrent/contrib/apachelog/referers.txt"); ApacheLogParseMapOutputOperator parse = dag.addOperator("parse", new ApacheLogParseMapOutputOperator()); GeoIPExtractor geoIPExtractor = new GeoIPExtractor(); geoIPExtractor.setDatabasePath("/home/david/GeoLiteCity.dat"); parse.registerInformationExtractor("ip", geoIPExtractor); parse.registerInformationExtractor("agent", new UserAgentExtractor()); TimestampExtractor timestampExtractor = new TimestampExtractor(); timestampExtractor.setDateFormatString("dd/MMM/yyyy:HH:mm:ss Z"); parse.registerInformationExtractor("time", timestampExtractor); ConsoleOutputOperator console = dag.addOperator("console", new ConsoleOutputOperator()); dag.addStream("log-parse", log.output, parse.data); dag.addStream("parse-console", parse.output, console.input).setLocality(Locality.CONTAINER_LOCAL); }  @Override void populateDAG(DAG dag, Configuration conf);  }
@Test public void testSomeMethod() throws Exception { ApplicationLogGenerator app = new ApplicationLogGenerator(); 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"); }