code: public static void main(String[] args) throws Exception {
  
  		// Checking input parameters
  		final ParameterTool params = ParameterTool.fromArgs(args);
  
  		StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
  		env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
  
  		DataStream<Integer> trainingData = env.addSource(new FiniteTrainingDataSource());
  		DataStream<Integer> newData = env.addSource(new FiniteNewDataSource());
  
  		// build new model on every second of new data
  		DataStream<Double[]> model = trainingData
  				.assignTimestampsAndWatermarks(new LinearTimestamp())
  				.timeWindowAll(Time.of(5000, TimeUnit.MILLISECONDS))
  				.apply(new PartialModelBuilder());
  
  		// use partial model for newData
  		DataStream<Integer> prediction = newData.connect(model).map(new Predictor());
  
  		// emit result
  		if (params.has("output")) {
  			prediction.writeAsText(params.get("output"));
  		} else {
  			System.out.println("Printing result to stdout. Use --output to specify output path.");
  			prediction.print();
  		}
  
  		// execute program
  		env.execute("Streaming Incremental Learning");
  	}
docs: *************************************************************************