@Test public void testLloyds() throws Exception { List<Centers> centers = ImmutableList.of( new Centers(ImmutableList.of(Vectors.of(1.0, 1.0), Vectors.of(5.0, 4.0)))); List<Centers> res = kmp.lloydsAlgorithm(VECS, centers, 0, false); assertEquals(centers, res); res = kmp.lloydsAlgorithm(VECS, res, 1, false); List<Centers> expected = ImmutableList.of( new Centers(ImmutableList.of(Vectors.of(1.5, 1.0), Vectors.of(4.5, 3.5)))); assertEquals(expected, res); res = kmp.lloydsAlgorithm(VECS, res, 1, false); assertEquals(expected, res); }
public <V extends RealVector> List<Centers> lloydsAlgorithm(PCollection<V> points, List<Centers> centers, int numIterations, boolean approx) { PTypeFamily tf = points.getTypeFamily(); PTableType<Pair<Integer, Integer>, Pair<V, Long>> ptt = tf.tableOf(tf.pairs(tf.ints(), tf.ints()), tf.pairs(points.getPType(), tf.longs())); Aggregator<Pair<V, Long>> agg = new SumVectorsAggregator<V>(); for (int i = 0; i < numIterations; i++) { KSketchIndex index = new KSketchIndex(centers, projectionBits, projectionSamples, seed); LloydsMapFn<V> mapFn = new LloydsMapFn<V>(index, approx); centers = new LloydsCenters<V>(points.parallelDo("lloyds-" + i, mapFn, ptt) .groupByKey() .combineValues(agg), centers.size()).getValue(); } return centers; }
KMeansParallel { public <V extends RealVector> List<Centers> lloydsAlgorithm(PCollection<V> points, List<Centers> centers, int numIterations, boolean approx) { PTypeFamily tf = points.getTypeFamily(); PTableType<Pair<Integer, Integer>, Pair<V, Long>> ptt = tf.tableOf(tf.pairs(tf.ints(), tf.ints()), tf.pairs(points.getPType(), tf.longs())); Aggregator<Pair<V, Long>> agg = new SumVectorsAggregator<V>(); for (int i = 0; i < numIterations; i++) { KSketchIndex index = new KSketchIndex(centers, projectionBits, projectionSamples, seed); LloydsMapFn<V> mapFn = new LloydsMapFn<V>(index, approx); centers = new LloydsCenters<V>(points.parallelDo("lloyds-" + i, mapFn, ptt) .groupByKey() .combineValues(agg), centers.size()).getValue(); } return centers; } }
KMeansParallel { public <V extends RealVector> List<Centers> lloydsAlgorithm(PCollection<V> points, List<Centers> centers, int numIterations, boolean approx) { PTypeFamily tf = points.getTypeFamily(); PTableType<Pair<Integer, Integer>, Pair<V, Long>> ptt = tf.tableOf(tf.pairs(tf.ints(), tf.ints()), tf.pairs(points.getPType(), tf.longs())); Aggregator<Pair<V, Long>> agg = new SumVectorsAggregator<V>(); for (int i = 0; i < numIterations; i++) { KSketchIndex index = new KSketchIndex(centers, projectionBits, projectionSamples, seed); LloydsMapFn<V> mapFn = new LloydsMapFn<V>(index, approx); centers = new LloydsCenters<V>(points.parallelDo("lloyds-" + i, mapFn, ptt) .groupByKey() .combineValues(agg), centers.size()).getValue(); } return centers; } KMeansParallel(RandomGenerator random, int projectionBits, int projectionSamples); }
KMeansParallel { public <V extends RealVector> List<Centers> lloydsAlgorithm(PCollection<V> points, List<Centers> centers, int numIterations, boolean approx) { PTypeFamily tf = points.getTypeFamily(); PTableType<Pair<Integer, Integer>, Pair<V, Long>> ptt = tf.tableOf(tf.pairs(tf.ints(), tf.ints()), tf.pairs(points.getPType(), tf.longs())); Aggregator<Pair<V, Long>> agg = new SumVectorsAggregator<V>(); for (int i = 0; i < numIterations; i++) { KSketchIndex index = new KSketchIndex(centers, projectionBits, projectionSamples, seed); LloydsMapFn<V> mapFn = new LloydsMapFn<V>(index, approx); centers = new LloydsCenters<V>(points.parallelDo("lloyds-" + i, mapFn, ptt) .groupByKey() .combineValues(agg), centers.size()).getValue(); } return centers; } KMeansParallel(RandomGenerator random, int projectionBits, int projectionSamples); List<Centers> lloydsAlgorithm(PCollection<V> points, List<Centers> centers,
      int numIterations, boolean approx); }
KMeansParallel { public <V extends RealVector> List<Centers> lloydsAlgorithm(PCollection<V> points, List<Centers> centers, int numIterations, boolean approx) { PTypeFamily tf = points.getTypeFamily(); PTableType<Pair<Integer, Integer>, Pair<V, Long>> ptt = tf.tableOf(tf.pairs(tf.ints(), tf.ints()), tf.pairs(points.getPType(), tf.longs())); Aggregator<Pair<V, Long>> agg = new SumVectorsAggregator<V>(); for (int i = 0; i < numIterations; i++) { KSketchIndex index = new KSketchIndex(centers, projectionBits, projectionSamples, seed); LloydsMapFn<V> mapFn = new LloydsMapFn<V>(index, approx); centers = new LloydsCenters<V>(points.parallelDo("lloyds-" + i, mapFn, ptt) .groupByKey() .combineValues(agg), centers.size()).getValue(); } return centers; } KMeansParallel(RandomGenerator random, int projectionBits, int projectionSamples); List<Centers> lloydsAlgorithm(PCollection<V> points, List<Centers> centers,
      int numIterations, boolean approx);  }