  protected void setup(Context context) throws IOException, InterruptedException {
    super.setup(context);
    Configuration conf = context.getConfiguration();
    try {
      ClassLoader ccl = Thread.currentThread().getContextClassLoader();
      DistanceMeasure measure = ccl.loadClass(conf.get(KMeansConfigKeys.DISTANCE_MEASURE_KEY))
          .asSubclass(DistanceMeasure.class).newInstance();
      measure.configure(conf);
      String clusterPath = conf.get(KMeansConfigKeys.CLUSTER_PATH_KEY);
      if ((clusterPath != null) && (clusterPath.length() > 0)) {
        KMeansUtil.configureWithClusterInfo(new Path(clusterPath), clusters);
        if (clusters.isEmpty()) {
          throw new IllegalStateException("Cluster is empty!");
        }
      }  
      this.clusterer = new KMeansClusterer(measure);
    } catch (ClassNotFoundException e) {
      throw new IllegalStateException(e);
    } catch (IllegalAccessException e) {
      throw new IllegalStateException(e);
    } catch (InstantiationException e) {
      throw new IllegalStateException(e);
    }
  }

  protected void setup(Context context) throws IOException, InterruptedException {
    super.setup(context);
    Configuration conf = context.getConfiguration();
    try {
      ClassLoader ccl = Thread.currentThread().getContextClassLoader();
      DistanceMeasure measure = ccl.loadClass(conf.get(KMeansConfigKeys.DISTANCE_MEASURE_KEY))
          .asSubclass(DistanceMeasure.class).newInstance();
      measure.configure(conf);
      String clusterPath = conf.get(KMeansConfigKeys.CLUSTER_PATH_KEY);
      if ((clusterPath != null) && (clusterPath.length() > 0)) {
        KMeansUtil.configureWithClusterInfo(new Path(clusterPath), clusters);
        if (clusters.isEmpty()) {
          throw new IllegalStateException("No clusters found. Check your -c path.");
        }
      }  
      this.clusterer = new KMeansClusterer(measure);
    } catch (ClassNotFoundException e) {
      throw new IllegalStateException(e);
    } catch (IllegalAccessException e) {
      throw new IllegalStateException(e);
    } catch (InstantiationException e) {
      throw new IllegalStateException(e);
    }
  }
