public static void main(String[] args) throws Exception {
		
		if(!parseParameters(args)) {
			return;
		}
	
		// set up execution environment
		ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
		
		// get input data
		DataSet<Point> points = getPointDataSet(env);
		DataSet<Centroid> centroids = getCentroidDataSet(env);
		
		// set number of bulk iterations for KMeans algorithm
		IterativeDataSet<Centroid> loop = centroids.iterate(numIterations);
		
		DataSet<Centroid> newCentroids = points
			// compute closest centroid for each point
			.map(new SelectNearestCenter()).withBroadcastSet(loop, "centroids")
			// count and sum point coordinates for each centroid
			.map(new CountAppender())
			.groupBy(0).reduce(new CentroidAccumulator())
			// compute new centroids from point counts and coordinate sums
			.map(new CentroidAverager());
		
		// feed new centroids back into next iteration
		DataSet<Centroid> finalCentroids = loop.closeWith(newCentroids);
		
		DataSet<Tuple2<Integer, Point>> clusteredPoints = points
				// assign points to final clusters
				.map(new SelectNearestCenter()).withBroadcastSet(finalCentroids, "centroids");
		
		// emit result
		if(fileOutput) {
			clusteredPoints.writeAsCsv(outputPath, "\n", " ");
		} else {
			clusteredPoints.print();
		}

		// execute program
		env.execute("KMeans Example");
		
	}
public static void main ( String [ ] args ) throws Exception { if ( ! parseParameters ( args ) ) { return ; } ExecutionEnvironment env = ExecutionEnvironment . getExecutionEnvironment ( ) ; DataSet < Point > points = getPointDataSet ( env ) ; DataSet < Centroid > centroids = getCentroidDataSet ( env ) ; IterativeDataSet < Centroid > loop = centroids . iterate ( numIterations ) ; DataSet < Centroid > newCentroids = points . map ( new SelectNearestCenter ( ) ) . withBroadcastSet ( loop , "centroids" ) . map ( new CountAppender ( ) ) . groupBy ( 0 ) . reduce ( new CentroidAccumulator ( ) ) . map ( new CentroidAverager ( ) ) ; DataSet < Centroid > finalCentroids = loop . closeWith ( newCentroids ) ; DataSet < Tuple2 < Integer , Point > > clusteredPoints = points . map ( new SelectNearestCenter ( ) ) . withBroadcastSet ( finalCentroids , "centroids" ) ; if ( fileOutput ) { clusteredPoints . writeAsCsv ( outputPath , "\n" , " " ) ; } else { clusteredPoints . print ( ) ; } env . execute ( "KMeans Example" ) ; }
*************************************************************************