code: public static void main(String[] args) throws Exception {
  		// parse the parameters
  		final ParameterTool params = ParameterTool.fromArgs(args);
  		final long windowSize = params.getLong("windowSize", 2000);
  		final long rate = params.getLong("rate", 3L);
  
  		System.out.println("Using windowSize=" + windowSize + ", data rate=" + rate);
  		System.out.println("To customize example, use: WindowJoin [--windowSize <window-size-in-millis>] [--rate <elements-per-second>]");
  
  		// obtain execution environment, run this example in "ingestion time"
  		StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
  		env.setStreamTimeCharacteristic(TimeCharacteristic.IngestionTime);
  
  		// make parameters available in the web interface
  		env.getConfig().setGlobalJobParameters(params);
  
  		// create the data sources for both grades and salaries
  		DataStream<Tuple2<String, Integer>> grades = GradeSource.getSource(env, rate);
  		DataStream<Tuple2<String, Integer>> salaries = SalarySource.getSource(env, rate);
  
  		// run the actual window join program
  		// for testability, this functionality is in a separate method.
  		DataStream<Tuple3<String, Integer, Integer>> joinedStream = runWindowJoin(grades, salaries, windowSize);
  
  		// print the results with a single thread, rather than in parallel
  		joinedStream.print().setParallelism(1);
  
  		// execute program
  		env.execute("Windowed Join Example");
  	}
docs: *************************************************************************