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");
}
 

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");
}
 

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");
}