code: public static void main(String... args) throws Exception {
  		
  		if(!parseParameters(args)) {
  			return;
  		}
  		
  		// set up execution environment
  		ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
  		
  		// read vertex and edge data
  		DataSet<Long> vertices = getVertexDataSet(env);
  		DataSet<Tuple2<Long, Long>> edges = getEdgeDataSet(env).flatMap(new UndirectEdge());
  		
  		// assign the initial components (equal to the vertex id)
  		DataSet<Tuple2<Long, Long>> verticesWithInitialId = vertices.map(new DuplicateValue<Long>());
  				
  		// open a delta iteration
  		DeltaIteration<Tuple2<Long, Long>, Tuple2<Long, Long>> iteration =
  				verticesWithInitialId.iterateDelta(verticesWithInitialId, maxIterations, 0);
  		
  		// apply the step logic: join with the edges, select the minimum neighbor, update if the component of the candidate is smaller
  		DataSet<Tuple2<Long, Long>> changes = iteration.getWorkset().join(edges).where(0).equalTo(0).with(new NeighborWithComponentIDJoin())
  				.groupBy(0).aggregate(Aggregations.MIN, 1)
  				.join(iteration.getSolutionSet()).where(0).equalTo(0)
  				.with(new ComponentIdFilter());
  
  		// close the delta iteration (delta and new workset are identical)
  		DataSet<Tuple2<Long, Long>> result = iteration.closeWith(changes, changes);
  		
  		// emit result
  		if(fileOutput) {
  			result.writeAsCsv(outputPath, "\n", " ");
  		} else {
  			result.print();
  		}
  				
  		// execute program
  		env.execute("Connected Components Example");
  				
  	}
docs: *************************************************************************