samoa-api/src/main/java/org/apache/samoa/moa/clusterers/clustream/Clustream.java - incubator-samoa - Git at Google

 package org.apache.samoa.moa.clusterers.clustream;

 /*
  * #%L
  * SAMOA
  * %%
  * Copyright (C) 2014 - 2015 Apache Software Foundation
  * %%
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
  * You may obtain a copy of the License at
  *
  *      http://www.apache.org/licenses/LICENSE-2.0
  *
  * Unless required by applicable law or agreed to in writing, software
  * distributed under the License is distributed on an "AS IS" BASIS,
  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  * See the License for the specific language governing permissions and
  * limitations under the License.
  * #L%
  */

 import java.util.ArrayList;
 import java.util.LinkedList;
 import java.util.List;
 import java.util.Random;

 import org.apache.samoa.instances.DenseInstance;
 import org.apache.samoa.instances.Instance;
 import org.apache.samoa.moa.cluster.Cluster;
 import org.apache.samoa.moa.cluster.Clustering;
 import org.apache.samoa.moa.cluster.SphereCluster;
 import org.apache.samoa.moa.clusterers.AbstractClusterer;
 import org.apache.samoa.moa.core.Measurement;

 import com.github.javacliparser.IntOption;

 /**
  * Citation: CluStream: Charu C. Aggarwal, Jiawei Han, Jianyong Wang, Philip S. Yu: A Framework for Clustering Evolving
  * Data Streams. VLDB 2003: 81-92
  */
 public class Clustream extends AbstractClusterer {

   private static final long serialVersionUID = 1L;

   public IntOption timeWindowOption = new IntOption("horizon",
       'h', "Rang of the window.", 1000);

   public IntOption maxNumKernelsOption = new IntOption(
       "maxNumKernels", 'k',
       "Maximum number of micro kernels to use.", 100);

   public IntOption kernelRadiFactorOption = new IntOption(
       "kernelRadiFactor", 't',
       "Multiplier for the kernel radius", 2);

   private int timeWindow;
   private long timestamp = -1;
   private ClustreamKernel[] kernels;
   private boolean initialized;
   private List<ClustreamKernel> buffer; // Buffer for initialization with kNN
   private int bufferSize;
   private double t;
   private int m;

   public Clustream() {
   }

   @Override
   public void resetLearningImpl() {
     this.kernels = new ClustreamKernel[maxNumKernelsOption.getValue()];
     this.timeWindow = timeWindowOption.getValue();
     this.initialized = false;
     this.buffer = new LinkedList<>();
     this.bufferSize = maxNumKernelsOption.getValue();
     t = kernelRadiFactorOption.getValue();
     m = maxNumKernelsOption.getValue();
   }

   @Override
   public void trainOnInstanceImpl(Instance instance) {
     int dim = instance.numValues();
     timestamp++;
     // 0. Initialize
     if (!initialized) {
       if (buffer.size() < bufferSize) {
         buffer.add(new ClustreamKernel(instance, dim, timestamp, t, m));
         return;
       }

       int k = kernels.length;
       // System.err.println("k="+k+" bufferSize="+bufferSize);
       assert (k <= bufferSize);

       ClustreamKernel[] centers = new ClustreamKernel[k];
       for (int i = 0; i < k; i++) {
         centers[i] = buffer.get(i); // TODO: make random!
       }
       Clustering kmeans_clustering = kMeans(k, centers, buffer);
       // Clustering kmeans_clustering = kMeans(k, buffer);

       for (int i = 0; i < kmeans_clustering.size(); i++) {
         kernels[i] = new ClustreamKernel(new DenseInstance(1.0, centers[i].getCenter()), dim, timestamp, t, m);
       }

       buffer.clear();
       initialized = true;
       return;
     }

     // 1. Determine closest kernel
     ClustreamKernel closestKernel = null;
     double minDistance = Double.MAX_VALUE;
     for (ClustreamKernel kernel : kernels) {
       // System.out.println(i+" "+kernels[i].getWeight()+" "+kernels[i].getDeviation());
       double distance = distance(instance.toDoubleArray(), kernel.getCenter());
       if (distance < minDistance) {
         closestKernel = kernel;
         minDistance = distance;
       }
     }

     // 2. Check whether instance fits into closestKernel
     double radius;
     if (closestKernel != null && closestKernel.getWeight() == 1) {
       // Special case: estimate radius by determining the distance to the
       // next closest cluster
       radius = Double.MAX_VALUE;
       double[] center = closestKernel.getCenter();
       for (ClustreamKernel kernel : kernels) {
         if (kernel == closestKernel) {
           continue;
         }

         double distance = distance(kernel.getCenter(), center);
         radius = Math.min(distance, radius);
       }
     } else {
       radius = closestKernel.getRadius();
     }

     if (minDistance < radius) {
       // Date fits, put into kernel and be happy
       closestKernel.insert(instance, timestamp);
       return;
     }

     // 3. Date does not fit, we need to free
     // some space to insert a new kernel
     long threshold = timestamp - timeWindow; // Kernels before this can be forgotten

     // 3.1 Try to forget old kernels
     for (int i = 0; i < kernels.length; i++) {
       if (kernels[i].getRelevanceStamp() < threshold) {
         kernels[i] = new ClustreamKernel(instance, dim, timestamp, t, m);
         return;
       }
     }

     // 3.2 Merge closest two kernels
     int closestA = 0;
     int closestB = 0;
     minDistance = Double.MAX_VALUE;
     for (int i = 0; i < kernels.length; i++) {
       double[] centerA = kernels[i].getCenter();
       for (int j = i + 1; j < kernels.length; j++) {
         double dist = distance(centerA, kernels[j].getCenter());
         if (dist < minDistance) {
           minDistance = dist;
           closestA = i;
           closestB = j;
         }
       }
     }
     assert (closestA != closestB);

     kernels[closestA].add(kernels[closestB]);
     kernels[closestB] = new ClustreamKernel(instance, dim, timestamp, t, m);
   }

   @Override
   public Clustering getMicroClusteringResult() {
     if (!initialized) {
       return new Clustering(new Cluster[0]);
     }

     ClustreamKernel[] res = new ClustreamKernel[kernels.length];
     for (int i = 0; i < res.length; i++) {
       res[i] = new ClustreamKernel(kernels[i], t, m);
     }

     return new Clustering(res);
   }

   @Override
   public boolean implementsMicroClusterer() {
     return true;
   }

   @Override
   public Clustering getClusteringResult() {
     return null;
   }

   public String getName() {
     return "Clustream " + timeWindow;
   }

   private static double distance(double[] pointA, double[] pointB) {
     double distance = 0.0;
     for (int i = 0; i < pointA.length; i++) {
       double d = pointA[i] - pointB[i];
       distance += d * d;
     }
     return Math.sqrt(distance);
   }

   // wrapper... we need to rewrite kmeans to points, not clusters, doesnt make
   // sense anymore
   // public static Clustering kMeans( int k, ArrayList<Instance> points, int dim
   // ) {
   // ArrayList<ClustreamKernel> cl = new ArrayList<ClustreamKernel>();
   // for(Instance inst : points){
   // cl.add(new ClustreamKernel(inst, dim , 0, 0, 0));
   // }
   // Clustering clustering = kMeans(k, cl);
   // return clustering;
   // }

   public static Clustering kMeans(int k, List<? extends Cluster> data) {
     Random random = new Random(0);
     Cluster[] centers = new Cluster[k];
     for (int i = 0; i < centers.length; i++) {
       int rid = random.nextInt(k);
       centers[i] = new SphereCluster(data.get(rid).getCenter(), 0);
     }
     return kMeans(k, centers, data);
   }

   public static Clustering kMeans(int k, Cluster[] centers, List<? extends Cluster> data) {
     assert (centers.length == k);
     assert (k > 0);

     int dimensions = centers[0].getCenter().length;

     ArrayList<ArrayList<Cluster>> clustering = new ArrayList<>();
     for (int i = 0; i < k; i++) {
       clustering.add(new ArrayList<Cluster>());
     }

     int repetitions = 100;
     while (repetitions-- >= 0) {
       // Assign points to clusters
       for (Cluster point : data) {
         double minDistance = distance(point.getCenter(), centers[0].getCenter());
         int closestCluster = 0;
         for (int i = 1; i < k; i++) {
           double distance = distance(point.getCenter(), centers[i].getCenter());
           if (distance < minDistance) {
             closestCluster = i;
             minDistance = distance;
           }
         }

         clustering.get(closestCluster).add(point);
       }

       // Calculate new centers and clear clustering lists
       SphereCluster[] newCenters = new SphereCluster[centers.length];
       for (int i = 0; i < k; i++) {
         newCenters[i] = calculateCenter(clustering.get(i), dimensions);
         clustering.get(i).clear();
       }
       centers = newCenters;
     }

     return new Clustering(centers);
   }

   private static SphereCluster calculateCenter(ArrayList<Cluster> cluster, int dimensions) {
     double[] res = new double[dimensions];
     for (int i = 0; i < res.length; i++) {
       res[i] = 0.0;
     }

     if (cluster.size() == 0) {
       return new SphereCluster(res, 0.0);
     }

     for (Cluster point : cluster) {
       double[] center = point.getCenter();
       for (int i = 0; i < res.length; i++) {
         res[i] += center[i];
       }
     }

     // Normalize
     for (int i = 0; i < res.length; i++) {
       res[i] /= cluster.size();
     }

     // Calculate radius
     double radius = 0.0;
     for (Cluster point : cluster) {
       double dist = distance(res, point.getCenter());
       if (dist > radius) {
         radius = dist;
       }
     }
     SphereCluster sc = new SphereCluster(res, radius);
     sc.setWeight(cluster.size());
     return sc;
   }

   @Override
   protected Measurement[] getModelMeasurementsImpl() {
     throw new UnsupportedOperationException("Not supported yet.");
   }

   @Override
   public void getModelDescription(StringBuilder out, int indent) {
     throw new UnsupportedOperationException("Not supported yet.");
   }

   public boolean isRandomizable() {
     return false;
   }

   public double[] getVotesForInstance(Instance inst) {
     throw new UnsupportedOperationException("Not supported yet.");
   }

 }
	package org.apache.samoa.moa.clusterers.clustream;

	/*
	* #%L
	* SAMOA
	* %%
	* Copyright (C) 2014 - 2015 Apache Software Foundation
	* %%
	* Licensed under the Apache License, Version 2.0 (the "License");
	* you may not use this file except in compliance with the License.
	* You may obtain a copy of the License at
	*
	* http://www.apache.org/licenses/LICENSE-2.0
	*
	* Unless required by applicable law or agreed to in writing, software
	* distributed under the License is distributed on an "AS IS" BASIS,
	* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
	* See the License for the specific language governing permissions and
	* limitations under the License.
	* #L%
	*/

	import java.util.ArrayList;
	import java.util.LinkedList;
	import java.util.List;
	import java.util.Random;

	import org.apache.samoa.instances.DenseInstance;
	import org.apache.samoa.instances.Instance;
	import org.apache.samoa.moa.cluster.Cluster;
	import org.apache.samoa.moa.cluster.Clustering;
	import org.apache.samoa.moa.cluster.SphereCluster;
	import org.apache.samoa.moa.clusterers.AbstractClusterer;
	import org.apache.samoa.moa.core.Measurement;

	import com.github.javacliparser.IntOption;

	/**
	* Citation: CluStream: Charu C. Aggarwal, Jiawei Han, Jianyong Wang, Philip S. Yu: A Framework for Clustering Evolving
	* Data Streams. VLDB 2003: 81-92
	*/
	public class Clustream extends AbstractClusterer {

	private static final long serialVersionUID = 1L;

	public IntOption timeWindowOption = new IntOption("horizon",
	'h', "Rang of the window.", 1000);

	public IntOption maxNumKernelsOption = new IntOption(
	"maxNumKernels", 'k',
	"Maximum number of micro kernels to use.", 100);

	public IntOption kernelRadiFactorOption = new IntOption(
	"kernelRadiFactor", 't',
	"Multiplier for the kernel radius", 2);

	private int timeWindow;
	private long timestamp = -1;
	private ClustreamKernel[] kernels;
	private boolean initialized;
	private List<ClustreamKernel> buffer; // Buffer for initialization with kNN
	private int bufferSize;
	private double t;
	private int m;

	public Clustream() {
	}

	@Override
	public void resetLearningImpl() {
	this.kernels = new ClustreamKernel[maxNumKernelsOption.getValue()];
	this.timeWindow = timeWindowOption.getValue();
	this.initialized = false;
	this.buffer = new LinkedList<>();
	this.bufferSize = maxNumKernelsOption.getValue();
	t = kernelRadiFactorOption.getValue();
	m = maxNumKernelsOption.getValue();
	}

	@Override
	public void trainOnInstanceImpl(Instance instance) {
	int dim = instance.numValues();
	timestamp++;
	// 0. Initialize
	if (!initialized) {
	if (buffer.size() < bufferSize) {
	buffer.add(new ClustreamKernel(instance, dim, timestamp, t, m));
	return;
	}

	int k = kernels.length;
	// System.err.println("k="+k+" bufferSize="+bufferSize);
	assert (k <= bufferSize);

	ClustreamKernel[] centers = new ClustreamKernel[k];
	for (int i = 0; i < k; i++) {
	centers[i] = buffer.get(i); // TODO: make random!
	}
	Clustering kmeans_clustering = kMeans(k, centers, buffer);
	// Clustering kmeans_clustering = kMeans(k, buffer);

	for (int i = 0; i < kmeans_clustering.size(); i++) {
	kernels[i] = new ClustreamKernel(new DenseInstance(1.0, centers[i].getCenter()), dim, timestamp, t, m);
	}

	buffer.clear();
	initialized = true;
	return;
	}

	// 1. Determine closest kernel
	ClustreamKernel closestKernel = null;
	double minDistance = Double.MAX_VALUE;
	for (ClustreamKernel kernel : kernels) {
	// System.out.println(i+" "+kernels[i].getWeight()+" "+kernels[i].getDeviation());
	double distance = distance(instance.toDoubleArray(), kernel.getCenter());
	if (distance < minDistance) {
	closestKernel = kernel;
	minDistance = distance;
	}
	}

	// 2. Check whether instance fits into closestKernel
	double radius;
	if (closestKernel != null && closestKernel.getWeight() == 1) {
	// Special case: estimate radius by determining the distance to the
	// next closest cluster
	radius = Double.MAX_VALUE;
	double[] center = closestKernel.getCenter();
	for (ClustreamKernel kernel : kernels) {
	if (kernel == closestKernel) {
	continue;
	}

	double distance = distance(kernel.getCenter(), center);
	radius = Math.min(distance, radius);
	}
	} else {
	radius = closestKernel.getRadius();
	}

	if (minDistance < radius) {
	// Date fits, put into kernel and be happy
	closestKernel.insert(instance, timestamp);
	return;
	}

	// 3. Date does not fit, we need to free
	// some space to insert a new kernel
	long threshold = timestamp - timeWindow; // Kernels before this can be forgotten

	// 3.1 Try to forget old kernels
	for (int i = 0; i < kernels.length; i++) {
	if (kernels[i].getRelevanceStamp() < threshold) {
	kernels[i] = new ClustreamKernel(instance, dim, timestamp, t, m);
	return;
	}
	}

	// 3.2 Merge closest two kernels
	int closestA = 0;
	int closestB = 0;
	minDistance = Double.MAX_VALUE;
	for (int i = 0; i < kernels.length; i++) {
	double[] centerA = kernels[i].getCenter();
	for (int j = i + 1; j < kernels.length; j++) {
	double dist = distance(centerA, kernels[j].getCenter());
	if (dist < minDistance) {
	minDistance = dist;
	closestA = i;
	closestB = j;
	}
	}
	}
	assert (closestA != closestB);

	kernels[closestA].add(kernels[closestB]);
	kernels[closestB] = new ClustreamKernel(instance, dim, timestamp, t, m);
	}

	@Override
	public Clustering getMicroClusteringResult() {
	if (!initialized) {
	return new Clustering(new Cluster[0]);
	}

	ClustreamKernel[] res = new ClustreamKernel[kernels.length];
	for (int i = 0; i < res.length; i++) {
	res[i] = new ClustreamKernel(kernels[i], t, m);
	}

	return new Clustering(res);
	}

	@Override
	public boolean implementsMicroClusterer() {
	return true;
	}

	@Override
	public Clustering getClusteringResult() {
	return null;
	}

	public String getName() {
	return "Clustream " + timeWindow;
	}

	private static double distance(double[] pointA, double[] pointB) {
	double distance = 0.0;
	for (int i = 0; i < pointA.length; i++) {
	double d = pointA[i] - pointB[i];
	distance += d * d;
	}
	return Math.sqrt(distance);
	}

	// wrapper... we need to rewrite kmeans to points, not clusters, doesnt make
	// sense anymore
	// public static Clustering kMeans( int k, ArrayList<Instance> points, int dim
	// ) {
	// ArrayList<ClustreamKernel> cl = new ArrayList<ClustreamKernel>();
	// for(Instance inst : points){
	// cl.add(new ClustreamKernel(inst, dim , 0, 0, 0));
	// }
	// Clustering clustering = kMeans(k, cl);
	// return clustering;
	// }

	public static Clustering kMeans(int k, List<? extends Cluster> data) {
	Random random = new Random(0);
	Cluster[] centers = new Cluster[k];
	for (int i = 0; i < centers.length; i++) {
	int rid = random.nextInt(k);
	centers[i] = new SphereCluster(data.get(rid).getCenter(), 0);
	}
	return kMeans(k, centers, data);
	}

	public static Clustering kMeans(int k, Cluster[] centers, List<? extends Cluster> data) {
	assert (centers.length == k);
	assert (k > 0);

	int dimensions = centers[0].getCenter().length;

	ArrayList<ArrayList<Cluster>> clustering = new ArrayList<>();
	for (int i = 0; i < k; i++) {
	clustering.add(new ArrayList<Cluster>());
	}

	int repetitions = 100;
	while (repetitions-- >= 0) {
	// Assign points to clusters
	for (Cluster point : data) {
	double minDistance = distance(point.getCenter(), centers[0].getCenter());
	int closestCluster = 0;
	for (int i = 1; i < k; i++) {
	double distance = distance(point.getCenter(), centers[i].getCenter());
	if (distance < minDistance) {
	closestCluster = i;
	minDistance = distance;
	}
	}

	clustering.get(closestCluster).add(point);
	}

	// Calculate new centers and clear clustering lists
	SphereCluster[] newCenters = new SphereCluster[centers.length];
	for (int i = 0; i < k; i++) {
	newCenters[i] = calculateCenter(clustering.get(i), dimensions);
	clustering.get(i).clear();
	}
	centers = newCenters;
	}

	return new Clustering(centers);
	}

	private static SphereCluster calculateCenter(ArrayList<Cluster> cluster, int dimensions) {
	double[] res = new double[dimensions];
	for (int i = 0; i < res.length; i++) {
	res[i] = 0.0;
	}

	if (cluster.size() == 0) {
	return new SphereCluster(res, 0.0);
	}

	for (Cluster point : cluster) {
	double[] center = point.getCenter();
	for (int i = 0; i < res.length; i++) {
	res[i] += center[i];
	}
	}

	// Normalize
	for (int i = 0; i < res.length; i++) {
	res[i] /= cluster.size();
	}

	// Calculate radius
	double radius = 0.0;
	for (Cluster point : cluster) {
	double dist = distance(res, point.getCenter());
	if (dist > radius) {
	radius = dist;
	}
	}
	SphereCluster sc = new SphereCluster(res, radius);
	sc.setWeight(cluster.size());
	return sc;
	}

	@Override
	protected Measurement[] getModelMeasurementsImpl() {
	throw new UnsupportedOperationException("Not supported yet.");
	}

	@Override
	public void getModelDescription(StringBuilder out, int indent) {
	throw new UnsupportedOperationException("Not supported yet.");
	}

	public boolean isRandomizable() {
	return false;
	}

	public double[] getVotesForInstance(Instance inst) {
	throw new UnsupportedOperationException("Not supported yet.");
	}

	}