docs: public static void main(String[] args) throws Exception {
  
  		ParameterTool params = ParameterTool.fromArgs(args);
  
  		final int numPages = params.getInt("numPages", PageRankData.getNumberOfPages());
  		final int maxIterations = params.getInt("iterations", 10);
  
  		// set up execution environment
  		final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
  
  		// make the parameters available to the web ui
  		env.getConfig().setGlobalJobParameters(params);
  
  		// get input data
  		DataSet<Long> pagesInput = getPagesDataSet(env, params);
  		DataSet<Tuple2<Long, Long>> linksInput = getLinksDataSet(env, params);
  
  		// assign initial rank to pages
  		DataSet<Tuple2<Long, Double>> pagesWithRanks = pagesInput.
  				map(new RankAssigner((1.0d / numPages)));
  
  		// build adjacency list from link input
  		DataSet<Tuple2<Long, Long[]>> adjacencyListInput =
  				linksInput.groupBy(0).reduceGroup(new BuildOutgoingEdgeList());
  
  		// set iterative data set
  		IterativeDataSet<Tuple2<Long, Double>> iteration = pagesWithRanks.iterate(maxIterations);
  
  		DataSet<Tuple2<Long, Double>> newRanks = iteration
  				// join pages with outgoing edges and distribute rank
  				.join(adjacencyListInput).where(0).equalTo(0).flatMap(new JoinVertexWithEdgesMatch())
  				// collect and sum ranks
  				.groupBy(0).aggregate(SUM, 1)
  				// apply dampening factor
  				.map(new Dampener(DAMPENING_FACTOR, numPages));
  
  		DataSet<Tuple2<Long, Double>> finalPageRanks = iteration.closeWith(
  				newRanks,
  				newRanks.join(iteration).where(0).equalTo(0)
  				// termination condition
  				.filter(new EpsilonFilter()));
  
  		// emit result
  		if (params.has("output")) {
  			finalPageRanks.writeAsCsv(params.get("output"), "\n", " ");
  			// execute program
  			env.execute("Basic Page Rank Example");
  		} else {
  			System.out.println("Printing result to stdout. Use --output to specify output path.");
  			finalPageRanks.print();
  		}
  	}