public static void main(String[] args) throws Exception {

		// set up execution environment
		StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
		StreamTableEnvironment tEnv = StreamTableEnvironment.create(env);

		DataStream<Order> orderA = env.fromCollection(Arrays.asList(
			new Order(1L, "beer", 3),
			new Order(1L, "diaper", 4),
			new Order(3L, "rubber", 2)));

		DataStream<Order> orderB = env.fromCollection(Arrays.asList(
			new Order(2L, "pen", 3),
			new Order(2L, "rubber", 3),
			new Order(4L, "beer", 1)));

		// convert DataStream to Table
		Table tableA = tEnv.fromDataStream(orderA, "user, product, amount");
		// register DataStream as Table
		tEnv.registerDataStream("OrderB", orderB, "user, product, amount");

		// union the two tables
		Table result = tEnv.sqlQuery("SELECT * FROM " + tableA + " WHERE amount > 2 UNION ALL " +
						"SELECT * FROM OrderB WHERE amount < 2");

		tEnv.toAppendStream(result, Order.class).print();

		env.execute();
	}
public static void main ( String [ ] args ) throws Exception { StreamExecutionEnvironment env = StreamExecutionEnvironment . getExecutionEnvironment ( ) ; StreamTableEnvironment tEnv = StreamTableEnvironment . create ( env ) ; DataStream < Order > orderA = env . fromCollection ( Arrays . asList ( new Order ( 1L , "beer" , 3 ) , new Order ( 1L , "diaper" , 4 ) , new Order ( 3L , "rubber" , 2 ) ) ) ; DataStream < Order > orderB = env . fromCollection ( Arrays . asList ( new Order ( 2L , "pen" , 3 ) , new Order ( 2L , "rubber" , 3 ) , new Order ( 4L , "beer" , 1 ) ) ) ; Table tableA = tEnv . fromDataStream ( orderA , "user, product, amount" ) ; tEnv . registerDataStream ( "OrderB" , orderB , "user, product, amount" ) ; Table result = tEnv . sqlQuery ( "SELECT * FROM " + tableA + " WHERE amount > 2 UNION ALL " + "SELECT * FROM OrderB WHERE amount < 2" ) ; tEnv . toAppendStream ( result , Order . class ) . print ( ) ; env . execute ( ) ; }
*************************************************************************