public void merge(StreamingHistogram other)
    {
        if (other == null)
            return;

        for (Map.Entry<Double, Long> entry : other.getAsMap().entrySet())
            update(entry.getKey(), entry.getValue());
    }

    /**
     * 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<Double, 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;
        }
        else
        {
            Map.Entry<Double, Long> pi = bin.floorEntry(b);
            if (pi == null)
                return 0;
            // calculate estimated count mb for point b
            double weight = (b - pi.getKey()) / (pnext.getKey() - pi.getKey());
            double mb = pi.getValue() + (pnext.getValue() - pi.getValue()) * weight;
            sum += (pi.getValue() + mb) * weight / 2;

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

    public Map<Double, Long> getAsMap()
    {
        return Collections.unmodifiableMap(bin);
    }

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

        public StreamingHistogram deserialize(DataInput in) throws IOException
        {
            int maxBinSize = in.readInt();
            int size = in.readInt();
            Map<Double, Long> tmp = new HashMap<>(size);
            for (int i = 0; i < size; i++)
            {
                tmp.put(in.readDouble(), in.readLong());
            }

            return new StreamingHistogram(maxBinSize, tmp);
        }

        public long serializedSize(StreamingHistogram histogram, TypeSizes typeSizes)
        {
            long size = typeSizes.sizeof(histogram.maxBinSize);
            Map<Double, Long> entries = histogram.getAsMap();
            size += typeSizes.sizeof(entries.size());
            // size of entries = size * (8(double) + 8(long))
            size += entries.size() * (8 + 8);
            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);
    }

}
public void merge ( StreamingHistogram other ) { if ( other == null ) return ; for ( Map . Entry < Double , Long > entry : other . getAsMap ( ) . entrySet ( ) ) update ( entry . getKey ( ) , entry . getValue ( ) ) ; } public double sum ( double b ) { double sum = 0 ; Map . Entry < Double , Long > pnext = bin . higherEntry ( b ) ; if ( pnext == null ) { for ( Long value : bin . values ( ) ) sum += value ; } else { Map . Entry < Double , Long > pi = bin . floorEntry ( b ) ; if ( pi == null ) return 0 ; double weight = ( b - pi . getKey ( ) ) / ( pnext . getKey ( ) - pi . getKey ( ) ) ; double mb = pi . getValue ( ) + ( pnext . getValue ( ) - pi . getValue ( ) ) * weight ; sum += ( pi . getValue ( ) + mb ) * weight / 2 ; sum += pi . getValue ( ) / 2.0 ; for ( Long value : bin . headMap ( pi . getKey ( ) , false ) . values ( ) ) sum += value ; } return sum ; } public Map < Double , Long > getAsMap ( ) { return Collections . unmodifiableMap ( bin ) ; } public static class StreamingHistogramSerializer implements ISerializer < StreamingHistogram > { public void serialize ( StreamingHistogram histogram , DataOutputPlus out ) throws IOException { out . writeInt ( histogram . maxBinSize ) ; Map < Double , Long > entries = histogram . getAsMap ( ) ; out . writeInt ( entries . size ( ) ) ; for ( Map . Entry < Double , Long > entry : entries . entrySet ( ) ) { out . writeDouble ( entry . getKey ( ) ) ; out . writeLong ( entry . getValue ( ) ) ; } } public StreamingHistogram deserialize ( DataInput in ) throws IOException { int maxBinSize = in . readInt ( ) ; int size = in . readInt ( ) ; Map < Double , Long > tmp = new HashMap < > ( size ) ; for ( int i = 0 ; i < size ; i ++ ) { tmp . put ( in . readDouble ( ) , in . readLong ( ) ) ; } return new StreamingHistogram ( maxBinSize , tmp ) ; } public long serializedSize ( StreamingHistogram histogram , TypeSizes typeSizes ) { long size = typeSizes . sizeof ( histogram . maxBinSize ) ; Map < Double , Long > entries = histogram . getAsMap ( ) ; size += typeSizes . sizeof ( entries . size ( ) ) ; size += entries . size ( ) * ( 8 + 8 ) ; 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 ) ; } }
Merges given histogram with this histogram.

@param other histogram to merge