k-means clustering implementation

[vasia]

Hello,

I would like to implement the k-means clustering algorithm for okapi and I would really appreciate your input regarding some implementation choices.

The algorithm partitions N data points (observations) into k clusters.
The standard algorithm is iterative and works as follows.

Input: data points of the form { pointID, coordinatesVector }
Output: { pointID, centerID } & coordinates of each of the k cluster centers. Cluster centers do not have to belong to the input points.
Initialization: randomly choose k points from input

In each iteration:

1. each data point is assigned to the cluster center which is closest to it, by means of euclidean distance
2. new cluster centers are recomputed, by calculating the arithmetic mean of the assigned points
   Convergence is reached when the positions of the cluster centers do not change.

In Giraph, each data point will correspond to a vertex and execute step 1. The positions of the k centers can be stored in an aggregator and updated by the Master.

The GPS paper describes a different version of the algorithm, where

1. cluster centers are randomly chosen in each iteration
2. distance from the centers is calculated based on shortest paths (edge weights)
3. convergence is reached when the edge cut is less than some threshold

Facebook seems to have followed a similar implementation.

In my view, it's better going for the standard implementation and once we have a stable implementation, maybe extend it to support edge-cut as converge criterion and/or other variations.
Let me know your thoughts!

[claudiomartella]

it appears to me like your proposed implementation, also similar to FB's is the right one. I'd go for it.

[dlogothetis]

I think Claudio is right. FB's approach is the same when it comes to iteratively updating centroids with the proposed approach.

The only difference is in how they decide to stop. In their scenario data points correspond to users and they measure the number of edges that cross clusters. Otherwise in the general case there are no edges really between data points.

The GPS algorithm also assumes an underlying graph and uses the edge weights to compute distances from centroids. And it uses the same way to decide when to stop.

[vasia]

Great, thank you! I'll proceed with the standard implementation then.

[claudiomartella]

does really facebook consider cut edges for the halting condition? i would be impressed, because they don't really look at the topology of the social graph with k-means, so how would they be minimising those? I think that the way they use k-means is not connected at all with graph processing, they just exploit the fast iterations of giraph.

[vasia]

That's what confused me as well.
I'm sorry, I didn't provide the correct link above. I was refering to this presentation.
In slide 7, it seems they don't consider the graph topology, while in slide 11 they seem to be using the edge cut metric.

[vasia]

Hello,
I have some doubts I hope you can help me with:

1. I have created a vector-of-vectors aggregator to store the cluster centers coordinates.
   If there are k centers and points have dimensions dim, then the aggregator's value is a vector of k vectors, each of size dim.
   Each vertex is supposed to aggregate only the value in the vector which corresponds to its assigned cluster center. I would like to do something like aggregate(Vector v, int index), where index is the position of the vector to be updated in the aggregator, instead of the provided aggregate(VectorOfVectors v). Is there a way to make a "point aggregation" like this?
2. Regarding the initialization phase, how to randomly select the starting cluster centers form the input points? Should I assume knowledge on the points distribution?

Thank you!

[dlogothetis]

1. I think if you have a VectorOfVectors aggregator, it will require much more work to do what you want. You basically will have to write an aggregator that will aggregate objects of the type {Vector,index}. Not exactly sure how it would work, but it seems possible.

An alternative, which I think is simpler, is just to have N aggregators, each for every centroid. You give them predefined names of the type C_1,..,C_N or something similar, and every vertex can refer to the corresponding aggregator (read its value or aggregate) using these names.

This may even be more efficient too if Giraph distributes the aggregators independently (not sure about this though).

2. Regarding the initialisation, it there are various ways, and they can have an impact on the performance of the algorithm. Maybe as a start we can pick them randomly from the input data point set.

Ideally, we should make it easy to "plug" different initialization methods. I imagine this will be coordinated by the MasterCompute, i.e. it decides when initialization has finished (it could take multiple steps) before it starts the main computation.

By the way, Giraph already provides some aggregators that handle arrays or matrices. Look under the org.apache.giraph.aggregators.matrix package. Okapi also has some Writable types that represent matrices that are basically extensions of the jblas types that implement the Writable interface. Look under ml.grafos.okapi.common.jblas. You could use these ones too. Write now there is actually only a matrix of floats, which should be enough, but it's also very easy to implement another type if you need it.

[claudiomartella]

I agree with dionysios, probably one aggregator for each centroid could be the way to go, so you can share the load of aggregators management across the workers, although only benchmarking would say a final word.

for initialisation, what i've seen being used is a random pick of k elements in the input dataset, but there can be refinements (e.g. basically making sure they are distributed as uniformly as possible).

[vasia]

Great, thank you for the advice! I'll go for the one aggregator per center, but I will also keep my current vector-of-vectors implementation in case we need it for benchmarking.

[vasia]

Hello,
I have another question, this time regarding the output.
Is a pointID -> clusterId mapping needed as output or just the cluster centers final coordinates?
Also, since I'm keeping the cluster centers coordinates in aggregators, how could I make their values available after job completion?
Thanks!

[dlogothetis]

Well the output of the algorithm is actually the clustering itself, so the point->cluster mapping will be ythe main output.

Now, outputting the cluster centers might be useful too for different purposes (e.g. visualization, debugging etc), so it'd be indeed good to output them too.

For this you can use Giraph's AggregatorWriter implementations. In general, an aggregator writer takes all aggregators and writes them to some output, with the behavior being customizable. Check under org.apache.giraph.aggregators. The TextAggregatorWriter is a good example, although you may want to write a new one if the format is not what you want.

[vasia]

Hey,

In my current implementation, the points (vertices) have their own coordinates as their value and do not update their state during supersteps. Rather, they compute the closest center and add their coordinates to the appropriate center aggregator.

So, I can see two ways of making the point->cluster mapping the main output:

1. extending the vertex value and also store the currently assigned center there, or
2. after the master has confirmed convergence, initiate a final superstep, during which, each point will update its value to the closest final center.

I'm not quite sure which approach will be more efficient, storing a larger value in each node or having one additional superstep. Option 1 will also complicate a bit the aggregation, since I will need to extract the coordinates from the value and then aggregate.
Any thoughts?
Thanks!

[claudiomartella]

my 2 cents is that you should not extend the vertex value with additional data. Ideally you'd compute the closest centroid during the execution of the outputformat, but i'm not sure you're able to access aggregator data there.
I think one additional superstep should be fine, it will not produce any communication, as it is just a simple iteration over the vertices.

[dlogothetis]

Both options are reasonable, although option 1 seems cleaner to me. You'll just need to make a kind of a composite value that has an array and a cluster id as fields. I don't think the overhead of option 1 is that big. Probably in the common case the dimension of the coordinates array is so large that one extra integer doesn't add much.

[vasia]

Hey,

I have created a simple test to check my implementation and I ran into the following issue:
How can I make an aggregator that extends DefaultImmutableClassesGiraphConfigurable read some parameters from the configuration?
When testing the aggregator separately, I can call setConf() on the object.
Is there a way to do something similar in a test using InternalVertexRunner, like this one?
Thanks!