Hi Subash, I would not fetch the data to the client, do the computation there, and send it back, just for the purpose of writing it to a file.
Either 1) pull the results to the client and write the file from there or 2) compute the outliers in the cluster. I did not study your code completely, but the two nested loops and the condition are a join for example. I would go for option 2, if possible. Best, Fabian 2016-02-10 13:07 GMT+01:00 subash basnet <yasub...@gmail.com>: > Hello Fabian, > > I use the collect() method to get the elements locally and perform > operations on that and return the result as a collection. The collection > result is converted to the DataSet in the calling method. > Below is the code of *findOutliers *method: > > public static List<Tuple3> findOutliers(DataSet<Tuple2<Integer, Point>> > clusteredPoints, > DataSet<Centroid> centroids) throws Exception { > List<Tuple3> finalElements = new ArrayList<Tuple3>(); > *List<Tuple2<Integer, Point>> elements = clusteredPoints.collect();* > * List<Centroid> centroidList = centroids.collect();* > List<Tuple3<Centroid, Tuple2<Integer, Point>, Double>> > elementsWithDistance = new ArrayList<Tuple3<Centroid, > Tuple2<Integer, Point>, Double>>(); > for (Centroid centroid : centroidList) { > elementsWithDistance = new ArrayList<Tuple3<Centroid, Tuple2<Integer, > Point>, Double>>(); > double totalDistance = 0; > int elementsCount = 0; > for (Tuple2<Integer, Point> e : elements) { > // compute distance > if (e.f0 == centroid.id) { > Tuple3<Centroid, Tuple2<Integer, Point>, Double> newElement = new > Tuple3<Centroid, > Tuple2<Integer, Point>, Double>(); > double distance = e.f1.euclideanDistance(centroid); > totalDistance += distance; > newElement.setFields(centroid, e, distance); > elementsWithDistance.add(newElement); > elementsCount++; > } > } > // finding mean > double mean = totalDistance / elementsCount; > double sdTotalDistanceSquare = 0; > for (Tuple3<Centroid, Tuple2<Integer, Point>, Double> elementWithDistance > : elementsWithDistance) { > double distanceSquare = Math.pow(mean - elementWithDistance.f2, 2); > sdTotalDistanceSquare += distanceSquare; > } > double sd = Math.sqrt(sdTotalDistanceSquare / elementsCount); > double upperlimit = mean + 2 * sd; > double lowerlimit = mean - 2 * sd; > Tuple3<Integer, Point, Boolean> newElement = new Tuple3<Integer, Point, > Boolean>();// true > // = > // outlier > for (Tuple3<Centroid, Tuple2<Integer, Point>, Double> elementWithDistance > : elementsWithDistance) { > newElement = new Tuple3<Integer, Point, Boolean>(); > if (elementWithDistance.f2 < lowerlimit || elementWithDistance.f2 > > upperlimit) { > // set as outlier > newElement.setFields(elementWithDistance.f1.f0, elementWithDistance.f1.f1, > true); > } else { > newElement.setFields(elementWithDistance.f1.f0, elementWithDistance.f1.f1, > false); > } > finalElements.add(newElement); > } > } > return finalElements; > } > > I have attached herewith the screenshot of my project structure and > KMeansOutlierDetection.java file for more clarity. > > > Best Regards, > Subash Basnet > > On Wed, Feb 10, 2016 at 12:26 PM, Fabian Hueske <fhue...@gmail.com> wrote: > >> [image: Boxbe] <https://www.boxbe.com/overview> This message is eligible >> for Automatic Cleanup! (fhue...@gmail.com) Add cleanup rule >> <https://www.boxbe.com/popup?url=https%3A%2F%2Fwww.boxbe.com%2Fcleanup%3Ftoken%3D8hdIOJf0i4083WeIB%252BQUXfS8djluXs0JekXPLuRpitIZdx1%252FAH%252BEFK2XXNXEBi6cnglpq9HBimim9%252FKCQ7UDLHnqGh6CGYAGY7zzpc82QTIAjEQM22%252FkBmdko8aAcxcD2P3ax587Jik%253D%26key%3DNO%252B%252BpxOTI6yOrzMtJK8863zNLUnk0hGhdxHIyLoWxck%253D&tc_serial=24326433442&tc_rand=1135108797&utm_source=stf&utm_medium=email&utm_campaign=ANNO_CLEANUP_ADD&utm_content=001> >> | More info >> <http://blog.boxbe.com/general/boxbe-automatic-cleanup?tc_serial=24326433442&tc_rand=1135108797&utm_source=stf&utm_medium=email&utm_campaign=ANNO_CLEANUP_ADD&utm_content=001> >> >> Hi Subash, >> >> how is findOutliers implemented? >> >> It might be that you mix-up local and cluster computation. All DataSets >> are processed in the cluster. Please note the following: >> - ExecutionEnvironment.fromCollection() transforms a client local >> connection into a DataSet by serializing it and sending it to the cluster. >> - DataSet.collect() transforms a DataSet into a collection and ships it >> back to the client. >> >> So, does findOutliers operate on the cluster or on the local client, >> i.e., does it work with DataSet and send the result back as a collection or >> does it first collect the results as collection and operate on these? >> >> Best, Fabian >> >> 2016-02-10 12:13 GMT+01:00 subash basnet <yasub...@gmail.com>: >> >>> Hello Stefano, >>> >>> Yeah the type casting worked, thank you. But not able to print the >>> Dataset to the file. >>> >>> The default below code which writes the KMeans points along with their >>> centroid numbers to the file works fine: >>> // 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"); >>> if (fileOutput) { >>> clusteredPoints.writeAsCsv(outputPath, "\n", " "); >>> // since file sinks are lazy, we trigger the execution explicitly >>> env.execute("KMeans Example"); >>> } >>> >>> But my modified code below to find outlier: >>> // 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"); >>> *DataSet<Tuple3> fElements = >>> env.fromCollection(findOutliers(clusteredPoints, finalCentroids));* >>> if (fileOutput) { >>> *fElements.writeAsCsv(outputPath, "\n", " ");* >>> // since file sinks are lazy, we trigger the execution explicitly >>> env.execute("KMeans Example"); >>> } >>> >>> It's not writing to the file, the *result *folder does not get created >>> inside kmeans folder where my centers, points file are located. I am only >>> able to print it to the console via *fElements.print();* >>> >>> Does it have something to do with *env.exectue("")*, which must be set >>> somewhere in the previous case but not in my case. >>> >>> >>> >>> Best Regards, >>> Subash Basnet >>> >>> >>> On Tue, Feb 9, 2016 at 6:29 PM, Stefano Baghino < >>> stefano.bagh...@radicalbit.io> wrote: >>> >>>> [image: Boxbe] <https://www.boxbe.com/overview> This message is >>>> eligible for Automatic Cleanup! (stefano.bagh...@radicalbit.io) Add >>>> cleanup rule >>>> <https://www.boxbe.com/popup?url=https%3A%2F%2Fwww.boxbe.com%2Fcleanup%3Ftoken%3D1lghJuQA8DeL%252BeQeu%252BXtjFS3Ln6XzfMngdzEeoXhxNL9D%252Fev2KZxlYVTG7zXzAOKqyTfuHHhyjFCeIEJhsuw2xSonJ%252Fz9ELZJGQHf2k5wgw88cHdkws1iTkY3LXpay0T6G30GCRRcKpUcUeyr6wyDBlPBPj1idLV%26key%3DzNLvLvtrNviObgkHecr87NQBUPN5j9wZMWIyBsSzzNM%253D&tc_serial=24317106750&tc_rand=543304732&utm_source=stf&utm_medium=email&utm_campaign=ANNO_CLEANUP_ADD&utm_content=001> >>>> | More info >>>> <http://blog.boxbe.com/general/boxbe-automatic-cleanup?tc_serial=24317106750&tc_rand=543304732&utm_source=stf&utm_medium=email&utm_campaign=ANNO_CLEANUP_ADD&utm_content=001> >>>> >>>> Assuming your EnvironmentContext is named `env` Simply call: >>>> >>>> DataSet<Tuple3<Integer, Point, Boolean>> fElements = env. >>>> *fromCollection*(finalElements); >>>> >>>> Does this help? >>>> >>>> On Tue, Feb 9, 2016 at 6:06 PM, subash basnet <yasub...@gmail.com> >>>> wrote: >>>> >>>>> Hello all, >>>>> >>>>> I have performed a modification in KMeans code to detect outliers. I >>>>> have printed the output in the console but I am not able to write it to >>>>> the >>>>> file using the given 'writeAsCsv' method. >>>>> The problem is I generate a list of tuples. >>>>> My List is: >>>>> List<Tuple3> finalElements = new ArrayList<Tuple3>(); >>>>> Following is the datatype of the elements added to the list: >>>>> Tuple3<Integer, Point, Boolean> newElement = new Tuple3<Integer, >>>>> Point, Boolean>(); >>>>> finalElements.add(newElement); >>>>> Now I am stuck on how to convert this 'finalElements' to >>>>> DataSet<Tuple3<Integer, Point, Boolean>> fElements, >>>>> so that I could use >>>>> fElements.writeAsCsv(outputPath, "\n"," "); >>>>> >>>>> Best Regards, >>>>> Subash Basnet >>>>> >>>> >>>> >>>> >>>> -- >>>> BR, >>>> Stefano Baghino >>>> >>>> Software Engineer @ Radicalbit >>>> >>>> >>> >> >> >