Imported Upstream version 0.0~git20150223.0.b965b61
Dmitry Smirnov
8 years ago
| 0 | # Perks for Go (golang.org) | |
| 1 | ||
| 2 | Perks contains the Go package quantile that computes approximate quantiles over | |
| 3 | an unbounded data stream within low memory and CPU bounds. | |
| 4 | ||
| 5 | For more information and examples, see: | |
| 6 | http://godoc.org/github.com/bmizerany/perks | |
| 7 | ||
| 8 | A very special thank you and shout out to Graham Cormode (Rutgers University), | |
| 9 | Flip Korn (AT&T Labs–Research), S. Muthukrishnan (Rutgers University), and | |
| 10 | Divesh Srivastava (AT&T Labs–Research) for their research and publication of | |
| 11 | [Effective Computation of Biased Quantiles over Data Streams](http://www.cs.rutgers.edu/~muthu/bquant.pdf) | |
| 12 | ||
| 13 | Thank you, also: | |
| 14 | * Armon Dadgar (@armon) | |
| 15 | * Andrew Gerrand (@nf) | |
| 16 | * Brad Fitzpatrick (@bradfitz) | |
| 17 | * Keith Rarick (@kr) | |
| 18 | ||
| 19 | FAQ: | |
| 20 | ||
| 21 | Q: Why not move the quantile package into the project root? | |
| 22 | A: I want to add more packages to perks later. | |
| 23 | ||
| 24 | Copyright (C) 2013 Blake Mizerany | |
| 25 | ||
| 26 | Permission is hereby granted, free of charge, to any person obtaining a copy of this software and associated documentation files (the "Software"), to deal in the Software without restriction, including without limitation the rights to use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the Software, and to permit persons to whom the Software is furnished to do so, subject to the following conditions: | |
| 27 | ||
| 28 | The above copyright notice and this permission notice shall be included in all copies or substantial portions of the Software. | |
| 29 | ||
| 30 | THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE. |
| 0 | package histogram | |
| 1 | ||
| 2 | import ( | |
| 3 | "math/rand" | |
| 4 | "testing" | |
| 5 | ) | |
| 6 | ||
| 7 | func BenchmarkInsert10Bins(b *testing.B) { | |
| 8 | b.StopTimer() | |
| 9 | h := New(10) | |
| 10 | b.StartTimer() | |
| 11 | for i := 0; i < b.N; i++ { | |
| 12 | f := rand.ExpFloat64() | |
| 13 | h.Insert(f) | |
| 14 | } | |
| 15 | } | |
| 16 | ||
| 17 | func BenchmarkInsert100Bins(b *testing.B) { | |
| 18 | b.StopTimer() | |
| 19 | h := New(100) | |
| 20 | b.StartTimer() | |
| 21 | for i := 0; i < b.N; i++ { | |
| 22 | f := rand.ExpFloat64() | |
| 23 | h.Insert(f) | |
| 24 | } | |
| 25 | } |
| 0 | // Package histogram provides a Go implementation of BigML's histogram package | |
| 1 | // for Clojure/Java. It is currently experimental. | |
| 2 | package histogram | |
| 3 | ||
| 4 | import ( | |
| 5 | "container/heap" | |
| 6 | "math" | |
| 7 | "sort" | |
| 8 | ) | |
| 9 | ||
| 10 | type Bin struct { | |
| 11 | Count int | |
| 12 | Sum float64 | |
| 13 | } | |
| 14 | ||
| 15 | func (b *Bin) Update(x *Bin) { | |
| 16 | b.Count += x.Count | |
| 17 | b.Sum += x.Sum | |
| 18 | } | |
| 19 | ||
| 20 | func (b *Bin) Mean() float64 { | |
| 21 | return b.Sum / float64(b.Count) | |
| 22 | } | |
| 23 | ||
| 24 | type Bins []*Bin | |
| 25 | ||
| 26 | func (bs Bins) Len() int { return len(bs) } | |
| 27 | func (bs Bins) Less(i, j int) bool { return bs[i].Mean() < bs[j].Mean() } | |
| 28 | func (bs Bins) Swap(i, j int) { bs[i], bs[j] = bs[j], bs[i] } | |
| 29 | ||
| 30 | func (bs *Bins) Push(x interface{}) { | |
| 31 | *bs = append(*bs, x.(*Bin)) | |
| 32 | } | |
| 33 | ||
| 34 | func (bs *Bins) Pop() interface{} { | |
| 35 | return bs.remove(len(*bs) - 1) | |
| 36 | } | |
| 37 | ||
| 38 | func (bs *Bins) remove(n int) *Bin { | |
| 39 | if n < 0 || len(*bs) < n { | |
| 40 | return nil | |
| 41 | } | |
| 42 | x := (*bs)[n] | |
| 43 | *bs = append((*bs)[:n], (*bs)[n+1:]...) | |
| 44 | return x | |
| 45 | } | |
| 46 | ||
| 47 | type Histogram struct { | |
| 48 | res *reservoir | |
| 49 | } | |
| 50 | ||
| 51 | func New(maxBins int) *Histogram { | |
| 52 | return &Histogram{res: newReservoir(maxBins)} | |
| 53 | } | |
| 54 | ||
| 55 | func (h *Histogram) Insert(f float64) { | |
| 56 | h.res.insert(&Bin{1, f}) | |
| 57 | h.res.compress() | |
| 58 | } | |
| 59 | ||
| 60 | func (h *Histogram) Bins() Bins { | |
| 61 | return h.res.bins | |
| 62 | } | |
| 63 | ||
| 64 | type reservoir struct { | |
| 65 | n int | |
| 66 | maxBins int | |
| 67 | bins Bins | |
| 68 | } | |
| 69 | ||
| 70 | func newReservoir(maxBins int) *reservoir { | |
| 71 | return &reservoir{maxBins: maxBins} | |
| 72 | } | |
| 73 | ||
| 74 | func (r *reservoir) insert(bin *Bin) { | |
| 75 | r.n += bin.Count | |
| 76 | i := sort.Search(len(r.bins), func(i int) bool { | |
| 77 | return r.bins[i].Mean() >= bin.Mean() | |
| 78 | }) | |
| 79 | if i < 0 || i == r.bins.Len() { | |
| 80 | // TODO(blake): Maybe use an .insert(i, bin) instead of | |
| 81 | // performing the extra work of a heap.Push. | |
| 82 | heap.Push(&r.bins, bin) | |
| 83 | return | |
| 84 | } | |
| 85 | r.bins[i].Update(bin) | |
| 86 | } | |
| 87 | ||
| 88 | func (r *reservoir) compress() { | |
| 89 | for r.bins.Len() > r.maxBins { | |
| 90 | minGapIndex := -1 | |
| 91 | minGap := math.MaxFloat64 | |
| 92 | for i := 0; i < r.bins.Len()-1; i++ { | |
| 93 | gap := gapWeight(r.bins[i], r.bins[i+1]) | |
| 94 | if minGap > gap { | |
| 95 | minGap = gap | |
| 96 | minGapIndex = i | |
| 97 | } | |
| 98 | } | |
| 99 | prev := r.bins[minGapIndex] | |
| 100 | next := r.bins.remove(minGapIndex + 1) | |
| 101 | prev.Update(next) | |
| 102 | } | |
| 103 | } | |
| 104 | ||
| 105 | func gapWeight(prev, next *Bin) float64 { | |
| 106 | return next.Mean() - prev.Mean() | |
| 107 | } |
| 0 | package histogram | |
| 1 | ||
| 2 | import ( | |
| 3 | "math/rand" | |
| 4 | "testing" | |
| 5 | ) | |
| 6 | ||
| 7 | func TestHistogram(t *testing.T) { | |
| 8 | const numPoints = 1e6 | |
| 9 | const maxBins = 3 | |
| 10 | ||
| 11 | h := New(maxBins) | |
| 12 | for i := 0; i < numPoints; i++ { | |
| 13 | f := rand.ExpFloat64() | |
| 14 | h.Insert(f) | |
| 15 | } | |
| 16 | ||
| 17 | bins := h.Bins() | |
| 18 | if g := len(bins); g > maxBins { | |
| 19 | t.Fatalf("got %d bins, wanted <= %d", g, maxBins) | |
| 20 | } | |
| 21 | ||
| 22 | for _, b := range bins { | |
| 23 | t.Logf("%+v", b) | |
| 24 | } | |
| 25 | ||
| 26 | if g := count(h.Bins()); g != numPoints { | |
| 27 | t.Fatalf("binned %d points, wanted %d", g, numPoints) | |
| 28 | } | |
| 29 | } | |
| 30 | ||
| 31 | func count(bins Bins) int { | |
| 32 | binCounts := 0 | |
| 33 | for _, b := range bins { | |
| 34 | binCounts += b.Count | |
| 35 | } | |
| 36 | return binCounts | |
| 37 | } |
| 0 | package quantile | |
| 1 | ||
| 2 | import ( | |
| 3 | "testing" | |
| 4 | ) | |
| 5 | ||
| 6 | func BenchmarkInsertTargeted(b *testing.B) { | |
| 7 | b.ReportAllocs() | |
| 8 | ||
| 9 | s := NewTargeted(Targets) | |
| 10 | b.ResetTimer() | |
| 11 | for i := float64(0); i < float64(b.N); i++ { | |
| 12 | s.Insert(i) | |
| 13 | } | |
| 14 | } | |
| 15 | ||
| 16 | func BenchmarkInsertTargetedSmallEpsilon(b *testing.B) { | |
| 17 | s := NewTargeted(TargetsSmallEpsilon) | |
| 18 | b.ResetTimer() | |
| 19 | for i := float64(0); i < float64(b.N); i++ { | |
| 20 | s.Insert(i) | |
| 21 | } | |
| 22 | } | |
| 23 | ||
| 24 | func BenchmarkInsertBiased(b *testing.B) { | |
| 25 | s := NewLowBiased(0.01) | |
| 26 | b.ResetTimer() | |
| 27 | for i := float64(0); i < float64(b.N); i++ { | |
| 28 | s.Insert(i) | |
| 29 | } | |
| 30 | } | |
| 31 | ||
| 32 | func BenchmarkInsertBiasedSmallEpsilon(b *testing.B) { | |
| 33 | s := NewLowBiased(0.0001) | |
| 34 | b.ResetTimer() | |
| 35 | for i := float64(0); i < float64(b.N); i++ { | |
| 36 | s.Insert(i) | |
| 37 | } | |
| 38 | } | |
| 39 | ||
| 40 | func BenchmarkQuery(b *testing.B) { | |
| 41 | s := NewTargeted(Targets) | |
| 42 | for i := float64(0); i < 1e6; i++ { | |
| 43 | s.Insert(i) | |
| 44 | } | |
| 45 | b.ResetTimer() | |
| 46 | n := float64(b.N) | |
| 47 | for i := float64(0); i < n; i++ { | |
| 48 | s.Query(i / n) | |
| 49 | } | |
| 50 | } | |
| 51 | ||
| 52 | func BenchmarkQuerySmallEpsilon(b *testing.B) { | |
| 53 | s := NewTargeted(TargetsSmallEpsilon) | |
| 54 | for i := float64(0); i < 1e6; i++ { | |
| 55 | s.Insert(i) | |
| 56 | } | |
| 57 | b.ResetTimer() | |
| 58 | n := float64(b.N) | |
| 59 | for i := float64(0); i < n; i++ { | |
| 60 | s.Query(i / n) | |
| 61 | } | |
| 62 | } |
| 0 | // +build go1.1 | |
| 1 | ||
| 2 | package quantile_test | |
| 3 | ||
| 4 | import ( | |
| 5 | "bufio" | |
| 6 | "fmt" | |
| 7 | "log" | |
| 8 | "os" | |
| 9 | "strconv" | |
| 10 | "time" | |
| 11 | ||
| 12 | "github.com/beorn7/perks/quantile" | |
| 13 | ) | |
| 14 | ||
| 15 | func Example_simple() { | |
| 16 | ch := make(chan float64) | |
| 17 | go sendFloats(ch) | |
| 18 | ||
| 19 | // Compute the 50th, 90th, and 99th percentile. | |
| 20 | q := quantile.NewTargeted(map[float64]float64{ | |
| 21 | 0.50: 0.005, | |
| 22 | 0.90: 0.001, | |
| 23 | 0.99: 0.0001, | |
| 24 | }) | |
| 25 | for v := range ch { | |
| 26 | q.Insert(v) | |
| 27 | } | |
| 28 | ||
| 29 | fmt.Println("perc50:", q.Query(0.50)) | |
| 30 | fmt.Println("perc90:", q.Query(0.90)) | |
| 31 | fmt.Println("perc99:", q.Query(0.99)) | |
| 32 | fmt.Println("count:", q.Count()) | |
| 33 | // Output: | |
| 34 | // perc50: 5 | |
| 35 | // perc90: 16 | |
| 36 | // perc99: 223 | |
| 37 | // count: 2388 | |
| 38 | } | |
| 39 | ||
| 40 | func Example_mergeMultipleStreams() { | |
| 41 | // Scenario: | |
| 42 | // We have multiple database shards. On each shard, there is a process | |
| 43 | // collecting query response times from the database logs and inserting | |
| 44 | // them into a Stream (created via NewTargeted(0.90)), much like the | |
| 45 | // Simple example. These processes expose a network interface for us to | |
| 46 | // ask them to serialize and send us the results of their | |
| 47 | // Stream.Samples so we may Merge and Query them. | |
| 48 | // | |
| 49 | // NOTES: | |
| 50 | // * These sample sets are small, allowing us to get them | |
| 51 | // across the network much faster than sending the entire list of data | |
| 52 | // points. | |
| 53 | // | |
| 54 | // * For this to work correctly, we must supply the same quantiles | |
| 55 | // a priori the process collecting the samples supplied to NewTargeted, | |
| 56 | // even if we do not plan to query them all here. | |
| 57 | ch := make(chan quantile.Samples) | |
| 58 | getDBQuerySamples(ch) | |
| 59 | q := quantile.NewTargeted(map[float64]float64{0.90: 0.001}) | |
| 60 | for samples := range ch { | |
| 61 | q.Merge(samples) | |
| 62 | } | |
| 63 | fmt.Println("perc90:", q.Query(0.90)) | |
| 64 | } | |
| 65 | ||
| 66 | func Example_window() { | |
| 67 | // Scenario: We want the 90th, 95th, and 99th percentiles for each | |
| 68 | // minute. | |
| 69 | ||
| 70 | ch := make(chan float64) | |
| 71 | go sendStreamValues(ch) | |
| 72 | ||
| 73 | tick := time.NewTicker(1 * time.Minute) | |
| 74 | q := quantile.NewTargeted(map[float64]float64{ | |
| 75 | 0.90: 0.001, | |
| 76 | 0.95: 0.0005, | |
| 77 | 0.99: 0.0001, | |
| 78 | }) | |
| 79 | for { | |
| 80 | select { | |
| 81 | case t := <-tick.C: | |
| 82 | flushToDB(t, q.Samples()) | |
| 83 | q.Reset() | |
| 84 | case v := <-ch: | |
| 85 | q.Insert(v) | |
| 86 | } | |
| 87 | } | |
| 88 | } | |
| 89 | ||
| 90 | func sendStreamValues(ch chan float64) { | |
| 91 | // Use your imagination | |
| 92 | } | |
| 93 | ||
| 94 | func flushToDB(t time.Time, samples quantile.Samples) { | |
| 95 | // Use your imagination | |
| 96 | } | |
| 97 | ||
| 98 | // This is a stub for the above example. In reality this would hit the remote | |
| 99 | // servers via http or something like it. | |
| 100 | func getDBQuerySamples(ch chan quantile.Samples) {} | |
| 101 | ||
| 102 | func sendFloats(ch chan<- float64) { | |
| 103 | f, err := os.Open("exampledata.txt") | |
| 104 | if err != nil { | |
| 105 | log.Fatal(err) | |
| 106 | } | |
| 107 | sc := bufio.NewScanner(f) | |
| 108 | for sc.Scan() { | |
| 109 | b := sc.Bytes() | |
| 110 | v, err := strconv.ParseFloat(string(b), 64) | |
| 111 | if err != nil { | |
| 112 | log.Fatal(err) | |
| 113 | } | |
| 114 | ch <- v | |
| 115 | } | |
| 116 | if sc.Err() != nil { | |
| 117 | log.Fatal(sc.Err()) | |
| 118 | } | |
| 119 | close(ch) | |
| 120 | } |
| 0 | // Package quantile computes approximate quantiles over an unbounded data | |
| 1 | // stream within low memory and CPU bounds. | |
| 2 | // | |
| 3 | // A small amount of accuracy is traded to achieve the above properties. | |
| 4 | // | |
| 5 | // Multiple streams can be merged before calling Query to generate a single set | |
| 6 | // of results. This is meaningful when the streams represent the same type of | |
| 7 | // data. See Merge and Samples. | |
| 8 | // | |
| 9 | // For more detailed information about the algorithm used, see: | |
| 10 | // | |
| 11 | // Effective Computation of Biased Quantiles over Data Streams | |
| 12 | // | |
| 13 | // http://www.cs.rutgers.edu/~muthu/bquant.pdf | |
| 14 | package quantile | |
| 15 | ||
| 16 | import ( | |
| 17 | "math" | |
| 18 | "sort" | |
| 19 | ) | |
| 20 | ||
| 21 | // Sample holds an observed value and meta information for compression. JSON | |
| 22 | // tags have been added for convenience. | |
| 23 | type Sample struct { | |
| 24 | Value float64 `json:",string"` | |
| 25 | Width float64 `json:",string"` | |
| 26 | Delta float64 `json:",string"` | |
| 27 | } | |
| 28 | ||
| 29 | // Samples represents a slice of samples. It implements sort.Interface. | |
| 30 | type Samples []Sample | |
| 31 | ||
| 32 | func (a Samples) Len() int { return len(a) } | |
| 33 | func (a Samples) Less(i, j int) bool { return a[i].Value < a[j].Value } | |
| 34 | func (a Samples) Swap(i, j int) { a[i], a[j] = a[j], a[i] } | |
| 35 | ||
| 36 | type invariant func(s *stream, r float64) float64 | |
| 37 | ||
| 38 | // NewLowBiased returns an initialized Stream for low-biased quantiles | |
| 39 | // (e.g. 0.01, 0.1, 0.5) where the needed quantiles are not known a priori, but | |
| 40 | // error guarantees can still be given even for the lower ranks of the data | |
| 41 | // distribution. | |
| 42 | // | |
| 43 | // The provided epsilon is a relative error, i.e. the true quantile of a value | |
| 44 | // returned by a query is guaranteed to be within (1±Epsilon)*Quantile. | |
| 45 | // | |
| 46 | // See http://www.cs.rutgers.edu/~muthu/bquant.pdf for time, space, and error | |
| 47 | // properties. | |
| 48 | func NewLowBiased(epsilon float64) *Stream { | |
| 49 | ƒ := func(s *stream, r float64) float64 { | |
| 50 | return 2 * epsilon * r | |
| 51 | } | |
| 52 | return newStream(ƒ) | |
| 53 | } | |
| 54 | ||
| 55 | // NewHighBiased returns an initialized Stream for high-biased quantiles | |
| 56 | // (e.g. 0.01, 0.1, 0.5) where the needed quantiles are not known a priori, but | |
| 57 | // error guarantees can still be given even for the higher ranks of the data | |
| 58 | // distribution. | |
| 59 | // | |
| 60 | // The provided epsilon is a relative error, i.e. the true quantile of a value | |
| 61 | // returned by a query is guaranteed to be within 1-(1±Epsilon)*(1-Quantile). | |
| 62 | // | |
| 63 | // See http://www.cs.rutgers.edu/~muthu/bquant.pdf for time, space, and error | |
| 64 | // properties. | |
| 65 | func NewHighBiased(epsilon float64) *Stream { | |
| 66 | ƒ := func(s *stream, r float64) float64 { | |
| 67 | return 2 * epsilon * (s.n - r) | |
| 68 | } | |
| 69 | return newStream(ƒ) | |
| 70 | } | |
| 71 | ||
| 72 | // NewTargeted returns an initialized Stream concerned with a particular set of | |
| 73 | // quantile values that are supplied a priori. Knowing these a priori reduces | |
| 74 | // space and computation time. The targets map maps the desired quantiles to | |
| 75 | // their absolute errors, i.e. the true quantile of a value returned by a query | |
| 76 | // is guaranteed to be within (Quantile±Epsilon). | |
| 77 | // | |
| 78 | // See http://www.cs.rutgers.edu/~muthu/bquant.pdf for time, space, and error properties. | |
| 79 | func NewTargeted(targets map[float64]float64) *Stream { | |
| 80 | ƒ := func(s *stream, r float64) float64 { | |
| 81 | var m = math.MaxFloat64 | |
| 82 | var f float64 | |
| 83 | for quantile, epsilon := range targets { | |
| 84 | if quantile*s.n <= r { | |
| 85 | f = (2 * epsilon * r) / quantile | |
| 86 | } else { | |
| 87 | f = (2 * epsilon * (s.n - r)) / (1 - quantile) | |
| 88 | } | |
| 89 | if f < m { | |
| 90 | m = f | |
| 91 | } | |
| 92 | } | |
| 93 | return m | |
| 94 | } | |
| 95 | return newStream(ƒ) | |
| 96 | } | |
| 97 | ||
| 98 | // Stream computes quantiles for a stream of float64s. It is not thread-safe by | |
| 99 | // design. Take care when using across multiple goroutines. | |
| 100 | type Stream struct { | |
| 101 | *stream | |
| 102 | b Samples | |
| 103 | sorted bool | |
| 104 | } | |
| 105 | ||
| 106 | func newStream(ƒ invariant) *Stream { | |
| 107 | x := &stream{ƒ: ƒ} | |
| 108 | return &Stream{x, make(Samples, 0, 500), true} | |
| 109 | } | |
| 110 | ||
| 111 | // Insert inserts v into the stream. | |
| 112 | func (s *Stream) Insert(v float64) { | |
| 113 | s.insert(Sample{Value: v, Width: 1}) | |
| 114 | } | |
| 115 | ||
| 116 | func (s *Stream) insert(sample Sample) { | |
| 117 | s.b = append(s.b, sample) | |
| 118 | s.sorted = false | |
| 119 | if len(s.b) == cap(s.b) { | |
| 120 | s.flush() | |
| 121 | } | |
| 122 | } | |
| 123 | ||
| 124 | // Query returns the computed qth percentiles value. If s was created with | |
| 125 | // NewTargeted, and q is not in the set of quantiles provided a priori, Query | |
| 126 | // will return an unspecified result. | |
| 127 | func (s *Stream) Query(q float64) float64 { | |
| 128 | if !s.flushed() { | |
| 129 | // Fast path when there hasn't been enough data for a flush; | |
| 130 | // this also yields better accuracy for small sets of data. | |
| 131 | l := len(s.b) | |
| 132 | if l == 0 { | |
| 133 | return 0 | |
| 134 | } | |
| 135 | i := int(float64(l) * q) | |
| 136 | if i > 0 { | |
| 137 | i -= 1 | |
| 138 | } | |
| 139 | s.maybeSort() | |
| 140 | return s.b[i].Value | |
| 141 | } | |
| 142 | s.flush() | |
| 143 | return s.stream.query(q) | |
| 144 | } | |
| 145 | ||
| 146 | // Merge merges samples into the underlying streams samples. This is handy when | |
| 147 | // merging multiple streams from separate threads, database shards, etc. | |
| 148 | // | |
| 149 | // ATTENTION: This method is broken and does not yield correct results. The | |
| 150 | // underlying algorithm is not capable of merging streams correctly. | |
| 151 | func (s *Stream) Merge(samples Samples) { | |
| 152 | sort.Sort(samples) | |
| 153 | s.stream.merge(samples) | |
| 154 | } | |
| 155 | ||
| 156 | // Reset reinitializes and clears the list reusing the samples buffer memory. | |
| 157 | func (s *Stream) Reset() { | |
| 158 | s.stream.reset() | |
| 159 | s.b = s.b[:0] | |
| 160 | } | |
| 161 | ||
| 162 | // Samples returns stream samples held by s. | |
| 163 | func (s *Stream) Samples() Samples { | |
| 164 | if !s.flushed() { | |
| 165 | return s.b | |
| 166 | } | |
| 167 | s.flush() | |
| 168 | return s.stream.samples() | |
| 169 | } | |
| 170 | ||
| 171 | // Count returns the total number of samples observed in the stream | |
| 172 | // since initialization. | |
| 173 | func (s *Stream) Count() int { | |
| 174 | return len(s.b) + s.stream.count() | |
| 175 | } | |
| 176 | ||
| 177 | func (s *Stream) flush() { | |
| 178 | s.maybeSort() | |
| 179 | s.stream.merge(s.b) | |
| 180 | s.b = s.b[:0] | |
| 181 | } | |
| 182 | ||
| 183 | func (s *Stream) maybeSort() { | |
| 184 | if !s.sorted { | |
| 185 | s.sorted = true | |
| 186 | sort.Sort(s.b) | |
| 187 | } | |
| 188 | } | |
| 189 | ||
| 190 | func (s *Stream) flushed() bool { | |
| 191 | return len(s.stream.l) > 0 | |
| 192 | } | |
| 193 | ||
| 194 | type stream struct { | |
| 195 | n float64 | |
| 196 | l []Sample | |
| 197 | ƒ invariant | |
| 198 | } | |
| 199 | ||
| 200 | func (s *stream) reset() { | |
| 201 | s.l = s.l[:0] | |
| 202 | s.n = 0 | |
| 203 | } | |
| 204 | ||
| 205 | func (s *stream) insert(v float64) { | |
| 206 | s.merge(Samples{{v, 1, 0}}) | |
| 207 | } | |
| 208 | ||
| 209 | func (s *stream) merge(samples Samples) { | |
| 210 | // TODO(beorn7): This tries to merge not only individual samples, but | |
| 211 | // whole summaries. The paper doesn't mention merging summaries at | |
| 212 | // all. Unittests show that the merging is inaccurate. Find out how to | |
| 213 | // do merges properly. | |
| 214 | var r float64 | |
| 215 | i := 0 | |
| 216 | for _, sample := range samples { | |
| 217 | for ; i < len(s.l); i++ { | |
| 218 | c := s.l[i] | |
| 219 | if c.Value > sample.Value { | |
| 220 | // Insert at position i. | |
| 221 | s.l = append(s.l, Sample{}) | |
| 222 | copy(s.l[i+1:], s.l[i:]) | |
| 223 | s.l[i] = Sample{ | |
| 224 | sample.Value, | |
| 225 | sample.Width, | |
| 226 | math.Max(sample.Delta, math.Floor(s.ƒ(s, r))-1), | |
| 227 | // TODO(beorn7): How to calculate delta correctly? | |
| 228 | } | |
| 229 | i++ | |
| 230 | goto inserted | |
| 231 | } | |
| 232 | r += c.Width | |
| 233 | } | |
| 234 | s.l = append(s.l, Sample{sample.Value, sample.Width, 0}) | |
| 235 | i++ | |
| 236 | inserted: | |
| 237 | s.n += sample.Width | |
| 238 | r += sample.Width | |
| 239 | } | |
| 240 | s.compress() | |
| 241 | } | |
| 242 | ||
| 243 | func (s *stream) count() int { | |
| 244 | return int(s.n) | |
| 245 | } | |
| 246 | ||
| 247 | func (s *stream) query(q float64) float64 { | |
| 248 | t := math.Ceil(q * s.n) | |
| 249 | t += math.Ceil(s.ƒ(s, t) / 2) | |
| 250 | p := s.l[0] | |
| 251 | var r float64 | |
| 252 | for _, c := range s.l[1:] { | |
| 253 | r += p.Width | |
| 254 | if r+c.Width+c.Delta > t { | |
| 255 | return p.Value | |
| 256 | } | |
| 257 | p = c | |
| 258 | } | |
| 259 | return p.Value | |
| 260 | } | |
| 261 | ||
| 262 | func (s *stream) compress() { | |
| 263 | if len(s.l) < 2 { | |
| 264 | return | |
| 265 | } | |
| 266 | x := s.l[len(s.l)-1] | |
| 267 | xi := len(s.l) - 1 | |
| 268 | r := s.n - 1 - x.Width | |
| 269 | ||
| 270 | for i := len(s.l) - 2; i >= 0; i-- { | |
| 271 | c := s.l[i] | |
| 272 | if c.Width+x.Width+x.Delta <= s.ƒ(s, r) { | |
| 273 | x.Width += c.Width | |
| 274 | s.l[xi] = x | |
| 275 | // Remove element at i. | |
| 276 | copy(s.l[i:], s.l[i+1:]) | |
| 277 | s.l = s.l[:len(s.l)-1] | |
| 278 | xi -= 1 | |
| 279 | } else { | |
| 280 | x = c | |
| 281 | xi = i | |
| 282 | } | |
| 283 | r -= c.Width | |
| 284 | } | |
| 285 | } | |
| 286 | ||
| 287 | func (s *stream) samples() Samples { | |
| 288 | samples := make(Samples, len(s.l)) | |
| 289 | copy(samples, s.l) | |
| 290 | return samples | |
| 291 | } |
| 0 | package quantile | |
| 1 | ||
| 2 | import ( | |
| 3 | "math" | |
| 4 | "math/rand" | |
| 5 | "sort" | |
| 6 | "testing" | |
| 7 | ) | |
| 8 | ||
| 9 | var ( | |
| 10 | Targets = map[float64]float64{ | |
| 11 | 0.01: 0.001, | |
| 12 | 0.10: 0.01, | |
| 13 | 0.50: 0.05, | |
| 14 | 0.90: 0.01, | |
| 15 | 0.99: 0.001, | |
| 16 | } | |
| 17 | TargetsSmallEpsilon = map[float64]float64{ | |
| 18 | 0.01: 0.0001, | |
| 19 | 0.10: 0.001, | |
| 20 | 0.50: 0.005, | |
| 21 | 0.90: 0.001, | |
| 22 | 0.99: 0.0001, | |
| 23 | } | |
| 24 | LowQuantiles = []float64{0.01, 0.1, 0.5} | |
| 25 | HighQuantiles = []float64{0.99, 0.9, 0.5} | |
| 26 | ) | |
| 27 | ||
| 28 | const RelativeEpsilon = 0.01 | |
| 29 | ||
| 30 | func verifyPercsWithAbsoluteEpsilon(t *testing.T, a []float64, s *Stream) { | |
| 31 | sort.Float64s(a) | |
| 32 | for quantile, epsilon := range Targets { | |
| 33 | n := float64(len(a)) | |
| 34 | k := int(quantile * n) | |
| 35 | lower := int((quantile - epsilon) * n) | |
| 36 | if lower < 1 { | |
| 37 | lower = 1 | |
| 38 | } | |
| 39 | upper := int(math.Ceil((quantile + epsilon) * n)) | |
| 40 | if upper > len(a) { | |
| 41 | upper = len(a) | |
| 42 | } | |
| 43 | w, min, max := a[k-1], a[lower-1], a[upper-1] | |
| 44 | if g := s.Query(quantile); g < min || g > max { | |
| 45 | t.Errorf("q=%f: want %v [%f,%f], got %v", quantile, w, min, max, g) | |
| 46 | } | |
| 47 | } | |
| 48 | } | |
| 49 | ||
| 50 | func verifyLowPercsWithRelativeEpsilon(t *testing.T, a []float64, s *Stream) { | |
| 51 | sort.Float64s(a) | |
| 52 | for _, qu := range LowQuantiles { | |
| 53 | n := float64(len(a)) | |
| 54 | k := int(qu * n) | |
| 55 | ||
| 56 | lowerRank := int((1 - RelativeEpsilon) * qu * n) | |
| 57 | upperRank := int(math.Ceil((1 + RelativeEpsilon) * qu * n)) | |
| 58 | w, min, max := a[k-1], a[lowerRank-1], a[upperRank-1] | |
| 59 | if g := s.Query(qu); g < min || g > max { | |
| 60 | t.Errorf("q=%f: want %v [%f,%f], got %v", qu, w, min, max, g) | |
| 61 | } | |
| 62 | } | |
| 63 | } | |
| 64 | ||
| 65 | func verifyHighPercsWithRelativeEpsilon(t *testing.T, a []float64, s *Stream) { | |
| 66 | sort.Float64s(a) | |
| 67 | for _, qu := range HighQuantiles { | |
| 68 | n := float64(len(a)) | |
| 69 | k := int(qu * n) | |
| 70 | ||
| 71 | lowerRank := int((1 - (1+RelativeEpsilon)*(1-qu)) * n) | |
| 72 | upperRank := int(math.Ceil((1 - (1-RelativeEpsilon)*(1-qu)) * n)) | |
| 73 | w, min, max := a[k-1], a[lowerRank-1], a[upperRank-1] | |
| 74 | if g := s.Query(qu); g < min || g > max { | |
| 75 | t.Errorf("q=%f: want %v [%f,%f], got %v", qu, w, min, max, g) | |
| 76 | } | |
| 77 | } | |
| 78 | } | |
| 79 | ||
| 80 | func populateStream(s *Stream) []float64 { | |
| 81 | a := make([]float64, 0, 1e5+100) | |
| 82 | for i := 0; i < cap(a); i++ { | |
| 83 | v := rand.NormFloat64() | |
| 84 | // Add 5% asymmetric outliers. | |
| 85 | if i%20 == 0 { | |
| 86 | v = v*v + 1 | |
| 87 | } | |
| 88 | s.Insert(v) | |
| 89 | a = append(a, v) | |
| 90 | } | |
| 91 | return a | |
| 92 | } | |
| 93 | ||
| 94 | func TestTargetedQuery(t *testing.T) { | |
| 95 | rand.Seed(42) | |
| 96 | s := NewTargeted(Targets) | |
| 97 | a := populateStream(s) | |
| 98 | verifyPercsWithAbsoluteEpsilon(t, a, s) | |
| 99 | } | |
| 100 | ||
| 101 | func TestLowBiasedQuery(t *testing.T) { | |
| 102 | rand.Seed(42) | |
| 103 | s := NewLowBiased(RelativeEpsilon) | |
| 104 | a := populateStream(s) | |
| 105 | verifyLowPercsWithRelativeEpsilon(t, a, s) | |
| 106 | } | |
| 107 | ||
| 108 | func TestHighBiasedQuery(t *testing.T) { | |
| 109 | rand.Seed(42) | |
| 110 | s := NewHighBiased(RelativeEpsilon) | |
| 111 | a := populateStream(s) | |
| 112 | verifyHighPercsWithRelativeEpsilon(t, a, s) | |
| 113 | } | |
| 114 | ||
| 115 | // BrokenTestTargetedMerge is broken, see Merge doc comment. | |
| 116 | func BrokenTestTargetedMerge(t *testing.T) { | |
| 117 | rand.Seed(42) | |
| 118 | s1 := NewTargeted(Targets) | |
| 119 | s2 := NewTargeted(Targets) | |
| 120 | a := populateStream(s1) | |
| 121 | a = append(a, populateStream(s2)...) | |
| 122 | s1.Merge(s2.Samples()) | |
| 123 | verifyPercsWithAbsoluteEpsilon(t, a, s1) | |
| 124 | } | |
| 125 | ||
| 126 | // BrokenTestLowBiasedMerge is broken, see Merge doc comment. | |
| 127 | func BrokenTestLowBiasedMerge(t *testing.T) { | |
| 128 | rand.Seed(42) | |
| 129 | s1 := NewLowBiased(RelativeEpsilon) | |
| 130 | s2 := NewLowBiased(RelativeEpsilon) | |
| 131 | a := populateStream(s1) | |
| 132 | a = append(a, populateStream(s2)...) | |
| 133 | s1.Merge(s2.Samples()) | |
| 134 | verifyLowPercsWithRelativeEpsilon(t, a, s2) | |
| 135 | } | |
| 136 | ||
| 137 | // BrokenTestHighBiasedMerge is broken, see Merge doc comment. | |
| 138 | func BrokenTestHighBiasedMerge(t *testing.T) { | |
| 139 | rand.Seed(42) | |
| 140 | s1 := NewHighBiased(RelativeEpsilon) | |
| 141 | s2 := NewHighBiased(RelativeEpsilon) | |
| 142 | a := populateStream(s1) | |
| 143 | a = append(a, populateStream(s2)...) | |
| 144 | s1.Merge(s2.Samples()) | |
| 145 | verifyHighPercsWithRelativeEpsilon(t, a, s2) | |
| 146 | } | |
| 147 | ||
| 148 | func TestUncompressed(t *testing.T) { | |
| 149 | q := NewTargeted(Targets) | |
| 150 | for i := 100; i > 0; i-- { | |
| 151 | q.Insert(float64(i)) | |
| 152 | } | |
| 153 | if g := q.Count(); g != 100 { | |
| 154 | t.Errorf("want count 100, got %d", g) | |
| 155 | } | |
| 156 | // Before compression, Query should have 100% accuracy. | |
| 157 | for quantile := range Targets { | |
| 158 | w := quantile * 100 | |
| 159 | if g := q.Query(quantile); g != w { | |
| 160 | t.Errorf("want %f, got %f", w, g) | |
| 161 | } | |
| 162 | } | |
| 163 | } | |
| 164 | ||
| 165 | func TestUncompressedSamples(t *testing.T) { | |
| 166 | q := NewTargeted(map[float64]float64{0.99: 0.001}) | |
| 167 | for i := 1; i <= 100; i++ { | |
| 168 | q.Insert(float64(i)) | |
| 169 | } | |
| 170 | if g := q.Samples().Len(); g != 100 { | |
| 171 | t.Errorf("want count 100, got %d", g) | |
| 172 | } | |
| 173 | } | |
| 174 | ||
| 175 | func TestUncompressedOne(t *testing.T) { | |
| 176 | q := NewTargeted(map[float64]float64{0.99: 0.01}) | |
| 177 | q.Insert(3.14) | |
| 178 | if g := q.Query(0.90); g != 3.14 { | |
| 179 | t.Error("want PI, got", g) | |
| 180 | } | |
| 181 | } | |
| 182 | ||
| 183 | func TestDefaults(t *testing.T) { | |
| 184 | if g := NewTargeted(map[float64]float64{0.99: 0.001}).Query(0.99); g != 0 { | |
| 185 | t.Errorf("want 0, got %f", g) | |
| 186 | } | |
| 187 | } |
| 0 | package topk | |
| 1 | ||
| 2 | import ( | |
| 3 | "sort" | |
| 4 | ) | |
| 5 | ||
| 6 | // http://www.cs.ucsb.edu/research/tech_reports/reports/2005-23.pdf | |
| 7 | ||
| 8 | type Element struct { | |
| 9 | Value string | |
| 10 | Count int | |
| 11 | } | |
| 12 | ||
| 13 | type Samples []*Element | |
| 14 | ||
| 15 | func (sm Samples) Len() int { | |
| 16 | return len(sm) | |
| 17 | } | |
| 18 | ||
| 19 | func (sm Samples) Less(i, j int) bool { | |
| 20 | return sm[i].Count < sm[j].Count | |
| 21 | } | |
| 22 | ||
| 23 | func (sm Samples) Swap(i, j int) { | |
| 24 | sm[i], sm[j] = sm[j], sm[i] | |
| 25 | } | |
| 26 | ||
| 27 | type Stream struct { | |
| 28 | k int | |
| 29 | mon map[string]*Element | |
| 30 | ||
| 31 | // the minimum Element | |
| 32 | min *Element | |
| 33 | } | |
| 34 | ||
| 35 | func New(k int) *Stream { | |
| 36 | s := new(Stream) | |
| 37 | s.k = k | |
| 38 | s.mon = make(map[string]*Element) | |
| 39 | s.min = &Element{} | |
| 40 | ||
| 41 | // Track k+1 so that less frequenet items contended for that spot, | |
| 42 | // resulting in k being more accurate. | |
| 43 | return s | |
| 44 | } | |
| 45 | ||
| 46 | func (s *Stream) Insert(x string) { | |
| 47 | s.insert(&Element{x, 1}) | |
| 48 | } | |
| 49 | ||
| 50 | func (s *Stream) Merge(sm Samples) { | |
| 51 | for _, e := range sm { | |
| 52 | s.insert(e) | |
| 53 | } | |
| 54 | } | |
| 55 | ||
| 56 | func (s *Stream) insert(in *Element) { | |
| 57 | e := s.mon[in.Value] | |
| 58 | if e != nil { | |
| 59 | e.Count++ | |
| 60 | } else { | |
| 61 | if len(s.mon) < s.k+1 { | |
| 62 | e = &Element{in.Value, in.Count} | |
| 63 | s.mon[in.Value] = e | |
| 64 | } else { | |
| 65 | e = s.min | |
| 66 | delete(s.mon, e.Value) | |
| 67 | e.Value = in.Value | |
| 68 | e.Count += in.Count | |
| 69 | s.min = e | |
| 70 | } | |
| 71 | } | |
| 72 | if e.Count < s.min.Count { | |
| 73 | s.min = e | |
| 74 | } | |
| 75 | } | |
| 76 | ||
| 77 | func (s *Stream) Query() Samples { | |
| 78 | var sm Samples | |
| 79 | for _, e := range s.mon { | |
| 80 | sm = append(sm, e) | |
| 81 | } | |
| 82 | sort.Sort(sort.Reverse(sm)) | |
| 83 | ||
| 84 | if len(sm) < s.k { | |
| 85 | return sm | |
| 86 | } | |
| 87 | ||
| 88 | return sm[:s.k] | |
| 89 | } |
| 0 | package topk | |
| 1 | ||
| 2 | import ( | |
| 3 | "fmt" | |
| 4 | "math/rand" | |
| 5 | "sort" | |
| 6 | "testing" | |
| 7 | ) | |
| 8 | ||
| 9 | func TestTopK(t *testing.T) { | |
| 10 | stream := New(10) | |
| 11 | ss := []*Stream{New(10), New(10), New(10)} | |
| 12 | m := make(map[string]int) | |
| 13 | for _, s := range ss { | |
| 14 | for i := 0; i < 1e6; i++ { | |
| 15 | v := fmt.Sprintf("%x", int8(rand.ExpFloat64())) | |
| 16 | s.Insert(v) | |
| 17 | m[v]++ | |
| 18 | } | |
| 19 | stream.Merge(s.Query()) | |
| 20 | } | |
| 21 | ||
| 22 | var sm Samples | |
| 23 | for x, s := range m { | |
| 24 | sm = append(sm, &Element{x, s}) | |
| 25 | } | |
| 26 | sort.Sort(sort.Reverse(sm)) | |
| 27 | ||
| 28 | g := stream.Query() | |
| 29 | if len(g) != 10 { | |
| 30 | t.Fatalf("got %d, want 10", len(g)) | |
| 31 | } | |
| 32 | for i, e := range g { | |
| 33 | if sm[i].Value != e.Value { | |
| 34 | t.Errorf("at %d: want %q, got %q", i, sm[i].Value, e.Value) | |
| 35 | } | |
| 36 | } | |
| 37 | } | |
| 38 | ||
| 39 | func TestQuery(t *testing.T) { | |
| 40 | queryTests := []struct { | |
| 41 | value string | |
| 42 | expected int | |
| 43 | }{ | |
| 44 | {"a", 1}, | |
| 45 | {"b", 2}, | |
| 46 | {"c", 2}, | |
| 47 | } | |
| 48 | ||
| 49 | stream := New(2) | |
| 50 | for _, tt := range queryTests { | |
| 51 | stream.Insert(tt.value) | |
| 52 | if n := len(stream.Query()); n != tt.expected { | |
| 53 | t.Errorf("want %d, got %d", tt.expected, n) | |
| 54 | } | |
| 55 | } | |
| 56 | } |