docs: public static void main(String[] args) throws Exception {
  
  		final ParameterTool params = ParameterTool.fromArgs(args);
  
  		final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
  		env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
  		env.getConfig().setGlobalJobParameters(params);
  
  		@SuppressWarnings({"rawtypes", "serial"})
  		DataStream<Tuple4<Integer, Integer, Double, Long>> carData;
  		if (params.has("input")) {
  			carData = env.readTextFile(params.get("input")).map(new ParseCarData());
  		} else {
  			System.out.println("Executing TopSpeedWindowing example with default input data set.");
  			System.out.println("Use --input to specify file input.");
  			carData = env.addSource(CarSource.create(2));
  		}
  
  		int evictionSec = 10;
  		double triggerMeters = 50;
  		DataStream<Tuple4<Integer, Integer, Double, Long>> topSpeeds = carData
  				.assignTimestampsAndWatermarks(new CarTimestamp())
  				.keyBy(0)
  				.window(GlobalWindows.create())
  				.evictor(TimeEvictor.of(Time.of(evictionSec, TimeUnit.SECONDS)))
  				.trigger(DeltaTrigger.of(triggerMeters,
  						new DeltaFunction<Tuple4<Integer, Integer, Double, Long>>() {
  							private static final long serialVersionUID = 1L;
  
  							@Override
  							public double getDelta(
  									Tuple4<Integer, Integer, Double, Long> oldDataPoint,
  									Tuple4<Integer, Integer, Double, Long> newDataPoint) {
  								return newDataPoint.f2 - oldDataPoint.f2;
  							}
  						}, carData.getType().createSerializer(env.getConfig())))
  				.maxBy(1);
  
  		if (params.has("output")) {
  			topSpeeds.writeAsText(params.get("output"));
  		} else {
  			System.out.println("Printing result to stdout. Use --output to specify output path.");
  			topSpeeds.print();
  		}
  
  		env.execute("CarTopSpeedWindowingExample");
  	}