FrequentItems Tutorial

Mix.install([
  {:ex_data_sketch, "~> 0.10"}
])

Introduction

ExDataSketch.FrequentItems finds the top-K most frequent items in a stream -- "what are the trending search terms," "which endpoints get the most traffic" -- using the SpaceSaving algorithm: at most k counters, each tracking one candidate item plus a bounded overcount error. Unlike ExDataSketch.CMS, which answers "what's the count for this item" and requires you to already know which items to ask about, FrequentItems directly hands you the ranked list.

Sample data (cached locally)

cache_path =
  Path.join(System.tmp_dir!(), "ex_data_sketch_livebook_cache/frequent_items_sample.bin")

queries =
  if File.exists?(cache_path) do
    IO.puts("Loading cached sample data from #{cache_path}")
    cache_path |> File.read!() |> :erlang.binary_to_term()
  else
    IO.puts("Generating sample data (this takes a few seconds)...")

    # 1,000,000 search queries over 5,000 distinct terms, power-law
    # distributed -- a handful of terms dominate, most appear rarely.
    queries =
      for _ <- 1..1_000_000 do
        rank = trunc(:math.pow(:rand.uniform(), 3) * 4_999) + 1
        "query_#{rank}"
      end

    File.mkdir_p!(Path.dirname(cache_path))
    File.write!(cache_path, :erlang.term_to_binary(queries))
    queries
  end

true_counts = Enum.frequencies(queries)
IO.puts("#{length(queries)} queries across #{map_size(true_counts)} distinct terms")

Basic usage

alias ExDataSketch.FrequentItems

sketch = FrequentItems.new(k: 20) |> FrequentItems.update_many(queries)

sketch |> FrequentItems.top_k() |> Enum.take(5)

Each entry is a map with :item, :estimate, :error (max possible overcount), :lower, and :upper -- SpaceSaving's guarantee is that the true count always falls in [lower, upper]:

top_5 = FrequentItems.top_k(sketch) |> Enum.take(5)

for entry <- top_5 do
  true_count = Map.get(true_counts, entry.item, 0)
  in_bounds = entry.lower <= true_count and true_count <= entry.upper

  IO.puts(
    "#{entry.item}: estimate=#{entry.estimate}, true=#{true_count}, " <>
      "bounds=[#{entry.lower}, #{entry.upper}], true in bounds: #{in_bounds}"
  )
end

frequent/2: items above an absolute count threshold

frequent = FrequentItems.frequent(sketch, 50_000)
IO.puts("#{length(frequent)} terms with a guaranteed count >= 50,000")

Sizing: k trade-off

More counters means more of the long tail gets tracked accurately before being evicted:

for k <- [5, 20, 100] do
  s = FrequentItems.new(k: k) |> FrequentItems.update_many(queries)
  top_1 = s |> FrequentItems.top_k() |> hd()
  IO.puts("k=#{k} (#{FrequentItems.size_bytes(s)} bytes): top term error bound=#{top_1.error}")
end

Merging

half = div(length(queries), 2)
{first_half, second_half} = Enum.split(queries, half)

worker_a = FrequentItems.new(k: 20) |> FrequentItems.update_many(first_half)
worker_b = FrequentItems.new(k: 20) |> FrequentItems.update_many(second_half)

merged = FrequentItems.merge(worker_a, worker_b)
merged |> FrequentItems.top_k() |> Enum.take(3)

Serialization

binary = FrequentItems.serialize(sketch)
{:ok, restored} = FrequentItems.deserialize(binary)
IO.puts("Round-tripped top term: #{restored |> FrequentItems.top_k() |> hd() |> Map.get(:item)}")