code: public static void main(String[] args) throws Exception {
  		
  		if(!parseParameters(args)) {
  			return;
  		}
  		
  		// set up execution environment
  		final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
  	
  		// read input data
  		DataSet<Edge> edges = getEdgeDataSet(env);
  		
  		// project edges by vertex id
  		DataSet<Edge> edgesById = edges
  				.map(new EdgeByIdProjector());
  		
  		DataSet<Triad> triangles = edgesById
  				// build triads
  				.groupBy(Edge.V1).sortGroup(Edge.V2, Order.ASCENDING).reduceGroup(new TriadBuilder())
  				// filter triads
  				.join(edgesById).where(Triad.V2, Triad.V3).equalTo(Edge.V1, Edge.V2).with(new TriadFilter());
  
  		// emit result
  		if(fileOutput) {
  			triangles.writeAsCsv(outputPath, "\n", ",");
  		} else {
  			triangles.print();
  		}
  
  		// execute program
  		env.execute("Basic Triangle Enumeration Example");
  
  	}
docs: *************************************************************************