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 
import asyncio
from dataclasses import dataclass
 
import aiohttp

Bounded-concurrency HTTP fetching with asyncio

async concurrency semaphore
Intermediate 8 steps
python 
from pathlib import Path
from typing import Iterator
 
 

Filtering files with pathlib.glob

generators filesystem filtering
Intermediate 6 steps
python 
from flask import Blueprint, jsonify
from sqlalchemy import text
from sqlalchemy.exc import SQLAlchemyError
 

Building a health check endpoint in Flask

health-check blueprint error-handling
Intermediate 6 steps
python 
import pandas as pd
import numpy as np
 
 

Cleaning a customer DataFrame with pandas

data-cleaning regex normalization
Intermediate 9 steps
python 
from datetime import date, timedelta
from typing import Annotated
 
from fastapi import APIRouter, Depends, Query

Validating date ranges with FastAPI dependencies

dependency-injection validation pydantic
Intermediate 6 steps
python 
from collections.abc import MutableMapping
 
 
class CaseInsensitiveDict(MutableMapping):

Building a case-insensitive dict in Python

data structures abstract base classes dunder methods
Intermediate 8 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