code: public static <T> DataSet<Tuple2<Long, T>> zipWithUniqueId (DataSet <T> input) {
  
  		return input.mapPartition(new RichMapPartitionFunction<T, Tuple2<Long, T>>() {
  
  			long maxBitSize = getBitSize(Long.MAX_VALUE);
  			long shifter = 0;
  			long start = 0;
  			long taskId = 0;
  			long label = 0;
  
  			@Override
  			public void open(Configuration parameters) throws Exception {
  				super.open(parameters);
  				shifter = getBitSize(getRuntimeContext().getNumberOfParallelSubtasks() - 1);
  				taskId = getRuntimeContext().getIndexOfThisSubtask();
  			}
  
  			@Override
  			public void mapPartition(Iterable<T> values, Collector<Tuple2<Long, T>> out) throws Exception {
  				for (T value : values) {
  					label = (start << shifter) + taskId;
  
  					if (getBitSize(start) + shifter < maxBitSize) {
  						out.collect(new Tuple2<>(label, value));
  						start++;
  					} else {
  						throw new Exception("Exceeded Long value range while generating labels");
  					}
  				}
  			}
  		});
  	}
docs: Method that assigns a unique {@link Long} value to all elements in the input data set as described below.
  <ul>
  <li> a map function is applied to the input data set
  <li> each map task holds a counter c which is increased for each record
  <li> c is shifted by n bits where n = log2(number of parallel tasks)
  <li> to create a unique ID among all tasks, the task id is added to the counter
  <li> for each record, the resulting counter is collected
  </ul>
  
  @param input the input data set
  @return a data set of tuple 2 consisting of ids and initial values.