TwitterDumpApplication implements StreamingApplication { @Override public void populateDAG(DAG dag, Configuration conf) { TwitterSampleInput twitterStream = dag.addOperator("TweetSampler", new TwitterSampleInput()); Status2Database dbWriter = dag.addOperator("DatabaseWriter", new Status2Database()); dbWriter.getStore().setDatabaseDriver("com.mysql.jdbc.Driver"); dbWriter.getStore().setDatabaseUrl("jdbc:mysql: dbWriter.getStore().setConnectionProperties("user:twitter"); dag.addStream("Statuses", twitterStream.status, dbWriter.input).setLocality(Locality.CONTAINER_LOCAL); }  @Override void populateDAG(DAG dag, Configuration conf);  }
@Test public void testPopulateDAG() throws Exception { Configuration configuration = new Configuration(false); LocalMode lm = LocalMode.newInstance(); DAG prepareDAG = lm.prepareDAG(new TwitterDumpApplication(), configuration); DAG clonedDAG = lm.cloneDAG(); assertEquals("Serialization", prepareDAG, clonedDAG); }