src/java/org/apache/cassandra/utils/StreamingHistogram.java - cassandra - Git at Google

 /*
  * Licensed to the Apache Software Foundation (ASF) under one
  * or more contributor license agreements.  See the NOTICE file
  * distributed with this work for additional information
  * regarding copyright ownership.  The ASF licenses this file
  * to you 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.
  */
 package org.apache.cassandra.utils;

 import java.io.IOException;
 import java.util.*;

 import com.google.common.base.Objects;

 import org.apache.cassandra.db.TypeSizes;
 import org.apache.cassandra.io.ISerializer;
 import org.apache.cassandra.io.sstable.SSTable;
 import org.apache.cassandra.io.util.DataInputPlus;
 import org.apache.cassandra.io.util.DataOutputPlus;

 /**
  * Histogram that can be constructed from streaming of data.
  *
  * The algorithm is taken from following paper:
  * Yael Ben-Haim and Elad Tom-Tov, "A Streaming Parallel Decision Tree Algorithm" (2010)
  * http://jmlr.csail.mit.edu/papers/volume11/ben-haim10a/ben-haim10a.pdf
  */
 public class StreamingHistogram
 {
     public static final StreamingHistogramSerializer serializer = new StreamingHistogramSerializer();

     // TreeMap to hold bins of histogram.
     // The key is a numeric type so we can avoid boxing/unboxing streams of different key types
     // The value is a unboxed long array always of length == 1
     // Serialized Histograms always writes with double keys for backwards compatibility
     private final TreeMap<Number, long[]> bin;

     // maximum bin size for this histogram
     private final int maxBinSize;


     /**
      * Creates a new histogram with max bin size of maxBinSize
      * @param maxBinSize maximum number of bins this histogram can have
      * @param source the existing bins in map form
      */
     private StreamingHistogram(int maxBinSize, Map<Number, long[]> source)
     {
         this.maxBinSize = maxBinSize;
         this.bin = new TreeMap<>((o1, o2) -> {
             if (o1.getClass().equals(o2.getClass()))
                 return ((Comparable)o1).compareTo(o2);
             else
                 return Double.compare(o1.doubleValue(), o2.doubleValue());
         });
         for (Map.Entry<Number, long[]> entry : source.entrySet())
             this.bin.put(entry.getKey(), new long[]{entry.getValue()[0]});
     }

     /**
      * Calculates estimated number of points in interval [-inf,b].
      *
      * @param b upper bound of a interval to calculate sum
      * @return estimated number of points in a interval [-inf,b].
      */
     public double sum(double b)
     {
         double sum = 0;
         // find the points pi, pnext which satisfy pi <= b < pnext
         Map.Entry<Number, long[]> pnext = bin.higherEntry(b);
         if (pnext == null)
         {
             // if b is greater than any key in this histogram,
             // just count all appearance and return
             for (long[] value : bin.values())
                 sum += value[0];
         }
         else
         {
             Map.Entry<Number, long[]> pi = bin.floorEntry(b);
             if (pi == null)
                 return 0;
             // calculate estimated count mb for point b
             double weight = (b - pi.getKey().doubleValue()) / (pnext.getKey().doubleValue() - pi.getKey().doubleValue());
             double mb = pi.getValue()[0] + (pnext.getValue()[0] - pi.getValue()[0]) * weight;
             sum += (pi.getValue()[0] + mb) * weight / 2;

             sum += pi.getValue()[0] / 2.0;
             for (long[] value : bin.headMap(pi.getKey(), false).values())
                 sum += value[0];
         }
         return sum;
     }

     public Map<Number, long[]> getAsMap()
     {
         return Collections.unmodifiableMap(bin);
     }

     public static class StreamingHistogramBuilder
     {
         // TreeMap to hold bins of histogram.
         // The key is a numeric type so we can avoid boxing/unboxing streams of different key types
         // The value is a unboxed long array always of length == 1
         // Serialized Histograms always writes with double keys for backwards compatibility
         private final TreeMap<Number, long[]> bin;

         // Keep a second, larger buffer to spool data in, before finalizing it into `bin`
         private final TreeMap<Number, long[]> spool;

         // maximum bin size for this histogram
         private final int maxBinSize;

         // maximum size of the spool
         private final int maxSpoolSize;

         // voluntarily give up resolution for speed
         private final int roundSeconds;

         /**
          * Creates a new histogram with max bin size of maxBinSize
          * @param maxBinSize maximum number of bins this histogram can have
          */
         public StreamingHistogramBuilder(int maxBinSize, int maxSpoolSize, int roundSeconds)
         {
             this.maxBinSize = maxBinSize;
             this.maxSpoolSize = maxSpoolSize;
             this.roundSeconds = roundSeconds;
             bin = new TreeMap<>((o1, o2) -> {
                 if (o1.getClass().equals(o2.getClass()))
                     return ((Comparable)o1).compareTo(o2);
                 else
                     return Double.compare(o1.doubleValue(), o2.doubleValue());
             });
             spool = new TreeMap<>((o1, o2) -> {
                 if (o1.getClass().equals(o2.getClass()))
                     return ((Comparable)o1).compareTo(o2);
                 else
                     return Double.compare(o1.doubleValue(), o2.doubleValue());
             });

         }

         public StreamingHistogram build()
         {
             flushHistogram();
             return new StreamingHistogram(maxBinSize,  bin);
         }

         /**
          * Adds new point p to this histogram.
          * @param p
          */
         public void update(Number p)
         {
             update(p, 1L);
         }

         /**
          * Adds new point p with value m to this histogram.
          * @param p
          * @param m
          */
         public void update(Number p, long m)
         {
             Number d = p.longValue() % this.roundSeconds;
             if (d.longValue() > 0)
                 p =p.longValue() + (this.roundSeconds - d.longValue());

             long[] mi = spool.get(p);
             if (mi != null)
             {
                 // we found the same p so increment that counter
                 mi[0] += m;
             }
             else
             {
                 mi = new long[]{m};
                 spool.put(p, mi);
             }

             // If spool has overflowed, compact it
             if(spool.size() > maxSpoolSize)
                 flushHistogram();
         }


         /**
          * Drain the temporary spool into the final bins
          */
         public void flushHistogram()
         {
             if (spool.size() > 0)
             {
                 long[] spoolValue;
                 long[] binValue;

                 // Iterate over the spool, copying the value into the primary bin map
                 // and compacting that map as necessary
                 for (Map.Entry<Number, long[]> entry : spool.entrySet())
                 {
                     Number key = entry.getKey();
                     spoolValue = entry.getValue();
                     binValue = bin.get(key);

                     // If this value is already in the final histogram bins
                     // Simply increment and update, otherwise, insert a new long[1] value
                     if(binValue != null)
                     {
                         binValue[0] += spoolValue[0];
                         bin.put(key, binValue);
                     }
                     else
                     {
                         bin.put(key, new long[]{spoolValue[0]});
                     }

                     if (bin.size() > maxBinSize)
                     {
                         // find points p1, p2 which have smallest difference
                         Iterator<Number> keys = bin.keySet().iterator();
                         double p1 = keys.next().doubleValue();
                         double p2 = keys.next().doubleValue();
                         double smallestDiff = p2 - p1;
                         double q1 = p1, q2 = p2;
                         while (keys.hasNext())
                         {
                             p1 = p2;
                             p2 = keys.next().doubleValue();
                             double diff = p2 - p1;
                             if (diff < smallestDiff)
                             {
                                 smallestDiff = diff;
                                 q1 = p1;
                                 q2 = p2;
                             }
                         }
                         // merge those two
                         long[] a1 = bin.remove(q1);
                         long[] a2 = bin.remove(q2);
                         long k1 = a1[0];
                         long k2 = a2[0];

                         a1[0] += k2;
                         bin.put((q1 * k1 + q2 * k2) / (k1 + k2), a1);

                     }
                 }
                 spool.clear();
             }
         }

         /**
          * Merges given histogram with this histogram.
          *
          * @param other histogram to merge
          */
         public void merge(StreamingHistogram other)
         {
             if (other == null)
                 return;

             for (Map.Entry<Number, long[]> entry : other.getAsMap().entrySet())
                 update(entry.getKey(), entry.getValue()[0]);
         }
     }

     public static class StreamingHistogramSerializer implements ISerializer<StreamingHistogram>
     {
         public void serialize(StreamingHistogram histogram, DataOutputPlus out) throws IOException
         {
             out.writeInt(histogram.maxBinSize);
             Map<Number, long[]> entries = histogram.getAsMap();
             out.writeInt(entries.size());
             for (Map.Entry<Number, long[]> entry : entries.entrySet())
             {
                 out.writeDouble(entry.getKey().doubleValue());
                 out.writeLong(entry.getValue()[0]);
             }
         }

         public StreamingHistogram deserialize(DataInputPlus in) throws IOException
         {
             int maxBinSize = in.readInt();
             int size = in.readInt();
             Map<Number, long[]> tmp = new HashMap<>(size);
             for (int i = 0; i < size; i++)
             {
                 tmp.put(in.readDouble(), new long[]{in.readLong()});
             }

             return new StreamingHistogram(maxBinSize, tmp);
         }

         public long serializedSize(StreamingHistogram histogram)
         {
             long size = TypeSizes.sizeof(histogram.maxBinSize);
             Map<Number, long[]> entries = histogram.getAsMap();
             size += TypeSizes.sizeof(entries.size());
             // size of entries = size * (8(double) + 8(long))
             size += entries.size() * (8L + 8L);
             return size;
         }
     }

     @Override
     public boolean equals(Object o)
     {
         if (this == o)
             return true;

         if (!(o instanceof StreamingHistogram))
             return false;

         StreamingHistogram that = (StreamingHistogram) o;
         return maxBinSize == that.maxBinSize &&
                bin.equals(that.bin);
     }

     @Override
     public int hashCode()
     {
         return Objects.hashCode(bin.hashCode(), maxBinSize);
     }

 }
	/*
	* Licensed to the Apache Software Foundation (ASF) under one
	* or more contributor license agreements. See the NOTICE file
	* distributed with this work for additional information
	* regarding copyright ownership. The ASF licenses this file
	* to you 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.
	*/
	package org.apache.cassandra.utils;

	import java.io.IOException;
	import java.util.*;

	import com.google.common.base.Objects;

	import org.apache.cassandra.db.TypeSizes;
	import org.apache.cassandra.io.ISerializer;
	import org.apache.cassandra.io.sstable.SSTable;
	import org.apache.cassandra.io.util.DataInputPlus;
	import org.apache.cassandra.io.util.DataOutputPlus;

	/**
	* Histogram that can be constructed from streaming of data.
	*
	* The algorithm is taken from following paper:
	* Yael Ben-Haim and Elad Tom-Tov, "A Streaming Parallel Decision Tree Algorithm" (2010)
	* http://jmlr.csail.mit.edu/papers/volume11/ben-haim10a/ben-haim10a.pdf
	*/
	public class StreamingHistogram
	{
	public static final StreamingHistogramSerializer serializer = new StreamingHistogramSerializer();

	// TreeMap to hold bins of histogram.
	// The key is a numeric type so we can avoid boxing/unboxing streams of different key types
	// The value is a unboxed long array always of length == 1
	// Serialized Histograms always writes with double keys for backwards compatibility
	private final TreeMap<Number, long[]> bin;

	// maximum bin size for this histogram
	private final int maxBinSize;


	/**
	* Creates a new histogram with max bin size of maxBinSize
	* @param maxBinSize maximum number of bins this histogram can have
	* @param source the existing bins in map form
	*/
	private StreamingHistogram(int maxBinSize, Map<Number, long[]> source)
	{
	this.maxBinSize = maxBinSize;
	this.bin = new TreeMap<>((o1, o2) -> {
	if (o1.getClass().equals(o2.getClass()))
	return ((Comparable)o1).compareTo(o2);
	else
	return Double.compare(o1.doubleValue(), o2.doubleValue());
	});
	for (Map.Entry<Number, long[]> entry : source.entrySet())
	this.bin.put(entry.getKey(), new long[]{entry.getValue()[0]});
	}

	/**
	* Calculates estimated number of points in interval [-inf,b].
	*
	* @param b upper bound of a interval to calculate sum
	* @return estimated number of points in a interval [-inf,b].
	*/
	public double sum(double b)
	{
	double sum = 0;
	// find the points pi, pnext which satisfy pi <= b < pnext
	Map.Entry<Number, long[]> pnext = bin.higherEntry(b);
	if (pnext == null)
	{
	// if b is greater than any key in this histogram,
	// just count all appearance and return
	for (long[] value : bin.values())
	sum += value[0];
	}
	else
	{
	Map.Entry<Number, long[]> pi = bin.floorEntry(b);
	if (pi == null)
	return 0;
	// calculate estimated count mb for point b
	double weight = (b - pi.getKey().doubleValue()) / (pnext.getKey().doubleValue() - pi.getKey().doubleValue());
	double mb = pi.getValue()[0] + (pnext.getValue()[0] - pi.getValue()[0]) * weight;
	sum += (pi.getValue()[0] + mb) * weight / 2;

	sum += pi.getValue()[0] / 2.0;
	for (long[] value : bin.headMap(pi.getKey(), false).values())
	sum += value[0];
	}
	return sum;
	}

	public Map<Number, long[]> getAsMap()
	{
	return Collections.unmodifiableMap(bin);
	}

	public static class StreamingHistogramBuilder
	{
	// TreeMap to hold bins of histogram.
	// The key is a numeric type so we can avoid boxing/unboxing streams of different key types
	// The value is a unboxed long array always of length == 1
	// Serialized Histograms always writes with double keys for backwards compatibility
	private final TreeMap<Number, long[]> bin;

	// Keep a second, larger buffer to spool data in, before finalizing it into `bin`
	private final TreeMap<Number, long[]> spool;

	// maximum bin size for this histogram
	private final int maxBinSize;

	// maximum size of the spool
	private final int maxSpoolSize;

	// voluntarily give up resolution for speed
	private final int roundSeconds;

	/**
	* Creates a new histogram with max bin size of maxBinSize
	* @param maxBinSize maximum number of bins this histogram can have
	*/
	public StreamingHistogramBuilder(int maxBinSize, int maxSpoolSize, int roundSeconds)
	{
	this.maxBinSize = maxBinSize;
	this.maxSpoolSize = maxSpoolSize;
	this.roundSeconds = roundSeconds;
	bin = new TreeMap<>((o1, o2) -> {
	if (o1.getClass().equals(o2.getClass()))
	return ((Comparable)o1).compareTo(o2);
	else
	return Double.compare(o1.doubleValue(), o2.doubleValue());
	});
	spool = new TreeMap<>((o1, o2) -> {
	if (o1.getClass().equals(o2.getClass()))
	return ((Comparable)o1).compareTo(o2);
	else
	return Double.compare(o1.doubleValue(), o2.doubleValue());
	});

	}

	public StreamingHistogram build()
	{
	flushHistogram();
	return new StreamingHistogram(maxBinSize, bin);
	}

	/**
	* Adds new point p to this histogram.
	* @param p
	*/
	public void update(Number p)
	{
	update(p, 1L);
	}

	/**
	* Adds new point p with value m to this histogram.
	* @param p
	* @param m
	*/
	public void update(Number p, long m)
	{
	Number d = p.longValue() % this.roundSeconds;
	if (d.longValue() > 0)
	p =p.longValue() + (this.roundSeconds - d.longValue());

	long[] mi = spool.get(p);
	if (mi != null)
	{
	// we found the same p so increment that counter
	mi[0] += m;
	}
	else
	{
	mi = new long[]{m};
	spool.put(p, mi);
	}

	// If spool has overflowed, compact it
	if(spool.size() > maxSpoolSize)
	flushHistogram();
	}



	/**
	* Drain the temporary spool into the final bins
	*/
	public void flushHistogram()
	{
	if (spool.size() > 0)
	{
	long[] spoolValue;
	long[] binValue;

	// Iterate over the spool, copying the value into the primary bin map
	// and compacting that map as necessary
	for (Map.Entry<Number, long[]> entry : spool.entrySet())
	{
	Number key = entry.getKey();
	spoolValue = entry.getValue();
	binValue = bin.get(key);

	// If this value is already in the final histogram bins
	// Simply increment and update, otherwise, insert a new long[1] value
	if(binValue != null)
	{
	binValue[0] += spoolValue[0];
	bin.put(key, binValue);
	}
	else
	{
	bin.put(key, new long[]{spoolValue[0]});
	}

	if (bin.size() > maxBinSize)
	{
	// find points p1, p2 which have smallest difference
	Iterator<Number> keys = bin.keySet().iterator();
	double p1 = keys.next().doubleValue();
	double p2 = keys.next().doubleValue();
	double smallestDiff = p2 - p1;
	double q1 = p1, q2 = p2;
	while (keys.hasNext())
	{
	p1 = p2;
	p2 = keys.next().doubleValue();
	double diff = p2 - p1;
	if (diff < smallestDiff)
	{
	smallestDiff = diff;
	q1 = p1;
	q2 = p2;
	}
	}
	// merge those two
	long[] a1 = bin.remove(q1);
	long[] a2 = bin.remove(q2);
	long k1 = a1[0];
	long k2 = a2[0];

	a1[0] += k2;
	bin.put((q1 * k1 + q2 * k2) / (k1 + k2), a1);

	}
	}
	spool.clear();
	}
	}

	/**
	* Merges given histogram with this histogram.
	*
	* @param other histogram to merge
	*/
	public void merge(StreamingHistogram other)
	{
	if (other == null)
	return;

	for (Map.Entry<Number, long[]> entry : other.getAsMap().entrySet())
	update(entry.getKey(), entry.getValue()[0]);
	}
	}

	public static class StreamingHistogramSerializer implements ISerializer<StreamingHistogram>
	{
	public void serialize(StreamingHistogram histogram, DataOutputPlus out) throws IOException
	{
	out.writeInt(histogram.maxBinSize);
	Map<Number, long[]> entries = histogram.getAsMap();
	out.writeInt(entries.size());
	for (Map.Entry<Number, long[]> entry : entries.entrySet())
	{
	out.writeDouble(entry.getKey().doubleValue());
	out.writeLong(entry.getValue()[0]);
	}
	}

	public StreamingHistogram deserialize(DataInputPlus in) throws IOException
	{
	int maxBinSize = in.readInt();
	int size = in.readInt();
	Map<Number, long[]> tmp = new HashMap<>(size);
	for (int i = 0; i < size; i++)
	{
	tmp.put(in.readDouble(), new long[]{in.readLong()});
	}

	return new StreamingHistogram(maxBinSize, tmp);
	}

	public long serializedSize(StreamingHistogram histogram)
	{
	long size = TypeSizes.sizeof(histogram.maxBinSize);
	Map<Number, long[]> entries = histogram.getAsMap();
	size += TypeSizes.sizeof(entries.size());
	// size of entries = size * (8(double) + 8(long))
	size += entries.size() * (8L + 8L);
	return size;
	}
	}

	@Override
	public boolean equals(Object o)
	{
	if (this == o)
	return true;

	if (!(o instanceof StreamingHistogram))
	return false;

	StreamingHistogram that = (StreamingHistogram) o;
	return maxBinSize == that.maxBinSize &&
	bin.equals(that.bin);
	}

	@Override
	public int hashCode()
	{
	return Objects.hashCode(bin.hashCode(), maxBinSize);
	}

	}