python 24 lines · 5 steps

Finding the top-N items in a stream

Count elements as they arrive, then use a heap to pull out the N most frequent without sorting everything.

Explained by highlit
Intermediate heaps counting generators streaming text-processing
1import heapq
2from collections import Counter
3from typing import Iterable, Hashable
4 
5 
6def top_n_from_stream(stream: Iterable[Hashable], n: int) -> list[tuple[Hashable, int]]:
7 counts: Counter[Hashable] = Counter()
8 for item in stream:
9 counts[item] += 1
10 
11 return heapq.nlargest(n, counts.items(), key=lambda pair: pair[1])
12 
13 
14def top_n_words(paths: Iterable[str], n: int = 10) -> list[tuple[str, int]]:
15 def token_stream() -> Iterable[str]:
16 for path in paths:
17 with open(path, encoding="utf-8") as fh:
18 for line in fh:
19 for token in line.split():
20 cleaned = token.strip(".,!?;:\"'()[]").lower()
21 if cleaned:
22 yield cleaned
23 
24 return top_n_from_stream(token_stream(), n)
01 / 01
STEP 01

Walkthrough

Space play ←→ step click any line
Three takeaways
  1. 1heapq.nlargest finds the top N in O(m log n) without fully sorting the collection.
  2. 2Generators let you process arbitrarily large input with constant memory over the token stream.
  3. 3Separating counting from top-N selection keeps each function small and reusable.

Related explainers

python 
from fastapi import FastAPI, WebSocket, WebSocketDisconnect
 
app = FastAPI()
 

Building a WebSocket chat with FastAPI

websockets broadcast connection-management
Intermediate 9 steps
python 
import time
import uuid
 
from django.utils.deprecation import MiddlewareMixin

Attaching per-request context in Django

middleware request lifecycle multi-tenancy
Intermediate 7 steps
python 
import random
from typing import Iterator, List
 
 

How reservoir sampling picks k items

reservoir-sampling streaming randomness
Intermediate 5 steps
python 
import secrets
 
from django.contrib.auth import authenticate, login
from django.core.cache import cache

Two-factor login with OTP in Django

two-factor-auth one-time-passwords caching
Intermediate 9 steps
ruby 
class SnippetHighlighter
  CONTEXT_RADIUS = 60
  MAX_TERMS = 8
 

Building search-result snippets in Ruby

regular-expressions text-processing search
Intermediate 9 steps
python 
import re
from functools import total_ordering
from typing import Optional
 

Parsing and comparing semantic versions

regex operator-overloading sorting
Intermediate 7 steps

Share this explainer

Here's the card — post it anywhere.

Finding the top-N items in a stream — share card
Made with highlit — turn any snippet into a walkthrough like this in about a minute.
Explain your code