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);
  		
  		// annotate edges with degrees
  		DataSet<EdgeWithDegrees> edgesWithDegrees = edges
  				.flatMap(new EdgeDuplicator())
  				.groupBy(Edge.V1).sortGroup(Edge.V2, Order.ASCENDING).reduceGroup(new DegreeCounter())
  				.groupBy(EdgeWithDegrees.V1,EdgeWithDegrees.V2).reduce(new DegreeJoiner());
  		
  		// project edges by degrees
  		DataSet<Edge> edgesByDegree = edgesWithDegrees
  				.map(new EdgeByDegreeProjector());
  		// project edges by vertex id
  		DataSet<Edge> edgesById = edgesByDegree
  				.map(new EdgeByIdProjector());
  		
  		DataSet<Triad> triangles = edgesByDegree
  				// 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("Triangle Enumeration Example");
  		
  	}
docs: *************************************************************************