Application implements StreamingApplication { @Override public void populateDAG(DAG dag, Configuration entries) { RandomKeysGenerator randGen = dag.addOperator("randomgen", new RandomKeysGenerator()); UniqueCounter<Integer> uniqCount = dag.addOperator("uniqevalue", new UniqueCounter<Integer>()); MapToKeyHashValuePairConverter<Integer, Integer> converter = dag.addOperator("converter", new MapToKeyHashValuePairConverter()); uniqCount.setCumulative(false); dag.setAttribute(uniqCount, Context.OperatorContext.PARTITIONER, new StatelessPartitioner<UniqueCounter<Integer>>(3)); CountVerifier<Integer> verifier = dag.addOperator("verifier", new CountVerifier<Integer>()); StreamDuplicater<KeyHashValPair<Integer, Integer>> dup = dag.addOperator("dup", new StreamDuplicater<KeyHashValPair<Integer, Integer>>()); ConsoleOutputOperator output = dag.addOperator("output", new ConsoleOutputOperator()); ConsoleOutputOperator successOutput = dag.addOperator("successoutput", new ConsoleOutputOperator()); successOutput.setStringFormat("Success %d"); ConsoleOutputOperator failureOutput = dag.addOperator("failureoutput", new ConsoleOutputOperator()); failureOutput.setStringFormat("Failure %d"); Counter successcounter = dag.addOperator("successcounter", new Counter()); Counter failurecounter = dag.addOperator("failurecounter", new Counter()); dag.addStream("datain", randGen.outPort, uniqCount.data); dag.addStream("dataverification0", randGen.verificationPort, verifier.in1); dag.addStream("convert", uniqCount.count, converter.input).setLocality(Locality.THREAD_LOCAL); dag.addStream("split", converter.output, dup.data); dag.addStream("consoutput", dup.out1, output.input); dag.addStream("dataverification1", dup.out2, verifier.in2); dag.addStream("successc", verifier.successPort, successcounter.input); dag.addStream("failurec", verifier.failurePort, failurecounter.input); dag.addStream("succconsoutput", successcounter.output, successOutput.input); dag.addStream("failconsoutput", failurecounter.output, failureOutput.input); }  @Override void populateDAG(DAG dag, Configuration entries);  }
@Test public void testApplication() throws Exception { LocalMode lma = LocalMode.newInstance(); new Application().populateDAG(lma.getDAG(), new Configuration(false)); LocalMode.Controller lc = lma.getController(); lc.run(10000); }