KMeansIterationsJobRunner implements MapReduceJobRunner, IndependentJobRunner { @SuppressWarnings("unchecked") @Override public int run(final Configuration config, final PropertyManagement runTimeProperties) throws Exception { convergenceTol = runTimeProperties.getPropertyAsDouble( ClusteringParameters.Clustering.CONVERGANCE_TOLERANCE, convergenceTol); final DistanceFn<T> distanceFunction = runTimeProperties.getClassInstance( CommonParameters.Common.DISTANCE_FUNCTION_CLASS, DistanceFn.class, FeatureCentroidDistanceFn.class); int maxIterationCount = runTimeProperties.getPropertyAsInt(ClusteringParameters.Clustering.MAX_ITERATIONS, 15); boolean converged = false; while (!converged && (maxIterationCount > 0)) { final int status = runJob(config, runTimeProperties); if (status != 0) { return status; } final CentroidManager<T> centroidManager = constructCentroidManager(config, runTimeProperties); converged = checkForConvergence(centroidManager, distanceFunction); maxIterationCount--; } return 0; } KMeansIterationsJobRunner(); void setInputFormatConfiguration(final FormatConfiguration inputFormatConfiguration); void setReducerCount(final int reducerCount); @SuppressWarnings("unchecked") @Override int run(final Configuration config, final PropertyManagement runTimeProperties); @Override Collection<ParameterEnum<?>> getParameters(); @Override int run(final PropertyManagement runTimeProperties);  }
@Test public void testRun() throws Exception { jobRunner.runJob(new Configuration(), propertyMgt); jobRunner.run(new Configuration(), propertyMgt); for (final Map.Entry<String, List<AnalyticItemWrapper<SimpleFeature>>> e : KMeansIterationsJobRunnerForTest.groups.entrySet()) { assertEquals(3, e.getValue().size()); for (final AnalyticItemWrapper<SimpleFeature> newCentroid : e.getValue()) { assertEquals(2, newCentroid.getIterationID()); boolean b = false; for (final AnalyticItemWrapper<SimpleFeature> oldCentroid : KMeansIterationsJobRunnerForTest.deletedSet.get( e.getKey())) { b |= oldCentroid.getID().equals(newCentroid.getID()); } assertFalse(b); } } for (final Map.Entry<String, List<AnalyticItemWrapper<SimpleFeature>>> e : KMeansIterationsJobRunnerForTest.deletedSet.entrySet()) { assertEquals(3, e.getValue().size()); for (final AnalyticItemWrapper<SimpleFeature> oldCentroid : e.getValue()) { assertEquals(1, oldCentroid.getIterationID()); } } }