topk

README

TopK: "filtered space saving" streaming topk algorithm

A modified version of http://github.com/dgryski/go-topk with the following changes

• Add msgp encoding/decoding
• Use metro hash
• Allow merging via https://ieeexplore.ieee.org/document/8438445

Documentation

Overview

Package topk implements the Filtered Space-Saving TopK streaming algorithm

The original Space-Saving algorithm: https://icmi.cs.ucsb.edu/research/tech_reports/reports/2005-23.pdf

The Filtered Space-Saving enhancement: http://www.l2f.inesc-id.pt/~fmmb/wiki/uploads/Work/misnis.ref0a.pdf

This implementation follows the algorithm of the FSS paper, but not the suggested implementation. Specifically, we use a heap instead of a sorted list of monitored items, and since we are also using a map to provide O(1) access on update also don't need the c_i counters in the hash table.

Licensed under the MIT license.

Index

• type Element
• type Stream
  • func (s *Stream) Decode(r io.Reader) error
  • func (s *Stream) DecodeMsgp(r *msgp.Reader) error
  • func (s *Stream) Encode(w io.Writer) error
  • func (s *Stream) EncodeMsgp(w *msgp.Writer) error
  • func (s *Stream) Estimate(x string) Element
  • func (s *Stream) Insert(x string, count int) Element
  • func (s *Stream) Keys() []Element
  • func (s *Stream) Merge(other *Stream) error
• type TopK
  • func New(k int) *TopK
  • func NewWithScaleFactor(k, m int) *TopK
  • func (t *TopK) Count() int
  • func (t *TopK) Decode(r io.Reader) error
  • func (t *TopK) DecodeMsgp(r *msgp.Reader) error
  • func (t *TopK) Encode(w io.Writer) error
  • func (t *TopK) EncodeMsgp(w *msgp.Writer) error
  • func (t *TopK) Insert(x string, count int) Element
  • func (t *TopK) Keys() []Element
  • func (t *TopK) Merge(other *TopK) error

Types

type Element

type Element struct {
	Key   string `json:"key"`
	Count int    `json:"count"`
	Error int    `json:"error"`
}

Element is a TopK item

type Stream

type Stream struct {
	// contains filtered or unexported fields
}

Stream calculates the TopK elements for a stream

func (*Stream) Decode

func (s *Stream) Decode(r io.Reader) error

Decode ...

func (*Stream) DecodeMsgp

func (s *Stream) DecodeMsgp(r *msgp.Reader) error

DecodeMsgp ...

func (*Stream) Encode

func (s *Stream) Encode(w io.Writer) error

Encode ...

func (*Stream) EncodeMsgp

func (s *Stream) EncodeMsgp(w *msgp.Writer) error

EncodeMsgp ...

func (*Stream) Estimate

func (s *Stream) Estimate(x string) Element

Estimate returns an estimate for the item x

func (*Stream) Insert

func (s *Stream) Insert(x string, count int) Element

Insert adds an element to the stream to be tracked It returns an estimation for the just inserted element

func (*Stream) Keys

func (s *Stream) Keys() []Element

Keys returns the current estimates for the most frequent elements

func (*Stream) Merge

func (s *Stream) Merge(other *Stream) error

Merge ...

type TopK

type TopK struct {
	*Stream
	// contains filtered or unexported fields
}

This is a simple wrapper around the Stream struct, more of a helper of sort

func New

func New(k int) *TopK

func NewWithScaleFactor

func NewWithScaleFactor(k, m int) *TopK

func (*TopK) Count

func (t *TopK) Count() int

Returns number of items inserted into the TopK

func (*TopK) Decode

func (t *TopK) Decode(r io.Reader) error

Decode ...

func (*TopK) DecodeMsgp

func (t *TopK) DecodeMsgp(r *msgp.Reader) error

DecodeMsgp ...

func (*TopK) Encode

func (t *TopK) Encode(w io.Writer) error

Encode ...

func (*TopK) EncodeMsgp

func (t *TopK) EncodeMsgp(w *msgp.Writer) error

EncodeMsgp ...

func (*TopK) Insert

func (t *TopK) Insert(x string, count int) Element

func (*TopK) Keys

func (t *TopK) Keys() []Element

func (*TopK) Merge

func (t *TopK) Merge(other *TopK) error