ApplicationWithDerbySQL implements StreamingApplication { @Override public void populateDAG(DAG dag, Configuration conf) { String symbolStr = conf.get(ApplicationWithDerbySQL.class.getName() + ".tickerSymbols", "AABA,GOOG,AAPL,FB,AMZN,NFLX,IBM"); String[] symbols = symbolStr.split(","); YahooFinanceCSVInputOperator input1 = dag.addOperator("input1", new YahooFinanceCSVInputOperator()); YahooFinanceCSVInputOperator input2 = dag.addOperator("input2", new YahooFinanceCSVInputOperator()); DerbySqlStreamOperator sqlOper = dag.addOperator("sqlOper", new DerbySqlStreamOperator()); ConsoleOutputOperator consoleOperator = dag.addOperator("console", new ConsoleOutputOperator()); for (String symbol : symbols) { input1.addSymbol(symbol); input2.addSymbol(symbol); } input1.addFormat("s0"); input1.addFormat("l1"); input2.addFormat("s0"); input2.addFormat("e0"); input2.addFormat("b4"); AbstractSqlStreamOperator.InputSchema inputSchema1 = new AbstractSqlStreamOperator.InputSchema("t1"); AbstractSqlStreamOperator.InputSchema inputSchema2 = new AbstractSqlStreamOperator.InputSchema("t2"); inputSchema1.setColumnInfo("s0", "varchar(100)", true); inputSchema1.setColumnInfo("l1", "float", false); inputSchema2.setColumnInfo("s0", "varchar(100)", true); inputSchema2.setColumnInfo("e0", "float", false); inputSchema2.setColumnInfo("b4", "float", false); sqlOper.setInputSchema(0, inputSchema1); sqlOper.setInputSchema(1, inputSchema2); sqlOper.addExecStatementString("SELECT SESSION.t1.s0 AS symbol, SESSION.t1.l1 / SESSION.t2.e0 AS pe_ratio, SESSION.t1.l1 / SESSION.t2.b4 AS pb_ratio FROM SESSION.t1,SESSION.t2 WHERE SESSION.t1.s0 = SESSION.t2.s0"); dag.addStream("input1_sql", input1.outputPort, sqlOper.in1); dag.addStream("input2_sql", input2.outputPort, sqlOper.in2); dag.addStream("result_console", sqlOper.result, consoleOperator.input); }  @Override void populateDAG(DAG dag, Configuration conf);  }
@Test public void testSomeMethod() throws Exception { LocalMode lma = LocalMode.newInstance(); new ApplicationWithDerbySQL().populateDAG(lma.getDAG(), new Configuration(false)); LocalMode.Controller lc = lma.getController(); long start = System.currentTimeMillis(); lc.run(); long end = System.currentTimeMillis(); long time = end - start; LOG.debug("Test used " + time + " ms"); }