code: public static void main(String[] args) throws Exception {
  
  		// Checking input parameters
  		final ParameterTool params = ParameterTool.fromArgs(args);
  
  		// set up execution environment
  		final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
  
  		// make parameters available in the web interface
  		env.getConfig().setGlobalJobParameters(params);
  
  		// read input data
  		DataSet<Edge> edges;
  		if (params.has("edges")) {
  			edges = env.readCsvFile(params.get("edges"))
  					.fieldDelimiter(" ")
  					.includeFields(true, true)
  					.types(Integer.class, Integer.class)
  					.map(new TupleEdgeConverter());
  		} else {
  			System.out.println("Executing EnumTriangles example with default edges data set.");
  			System.out.println("Use --edges to specify file input.");
  			edges = EnumTrianglesData.getDefaultEdgeDataSet(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 (params.has("output")) {
  			triangles.writeAsCsv(params.get("output"), "\n", ",");
  			// execute program
  			env.execute("Basic Triangle Enumeration Example");
  		} else {
  			System.out.println("Printing result to stdout. Use --output to specify output path.");
  			triangles.print();
  		}
  	}
docs: *************************************************************************