python
19 lines · 5 steps
How reservoir sampling picks k items
Draw a uniform random sample of k items from a stream of unknown length in a single pass.
Explained by
highlit
1import random
2from typing import Iterator, List
3
4
5def reservoir_sample(lines: Iterator[str], k: int) -> List[str]:
6 reservoir: List[str] = []
7 for i, line in enumerate(lines):
8 if i < k:
9 reservoir.append(line)
10 else:
11 j = random.randint(0, i)
12 if j < k:
13 reservoir[j] = line
14 return reservoir
15
16
17def sample_lines_from_file(path: str, k: int) -> List[str]:
18 with open(path, "r", encoding="utf-8") as fh:
19 return [line.rstrip("\n") for line in reservoir_sample(fh, k)]
01 / 01
STEP 01
‹ swipe to step through ›
Walkthrough
Space play
←→ step
click any line
Three takeaways
- 1Reservoir sampling gives a uniform sample without knowing the total count up front.
- 2Replacing element j with probability k/(i+1) keeps every seen item equally likely.
- 3Streaming over an iterator lets you sample files larger than memory in one pass.
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 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
python
import re from functools import total_ordering from typing import Optional
Parsing and comparing semantic versions
regex
operator-overloading
sorting
Intermediate
7 steps
python
from typing import Any, Sequence, Mapping def render_markdown_table(
Rendering an aligned Markdown table in Python
string formatting
data transformation
closures
Intermediate
8 steps
python
from django.db import connections from django.db.utils import OperationalError from django.core.cache import caches from django.http import JsonResponse
Building a health check endpoint in Django
health check
monitoring
database
Intermediate
7 steps
Share this explainer
Here's the card — post it anywhere.
Made with highlit — turn any snippet into a walkthrough like this in about a minute.
Explain your code
Embed this explainer
Drop the interactive walkthrough into a blog or docs. Views never cost a credit.
<iframe src="https://highlit.co/explainers/how-reservoir-sampling-picks-k-items-explained-python-0d4d/embed?autoplay=1" width="100%" height="520" loading="lazy" style="border:0"></iframe>
Autoplay is on by default — add ?autoplay=0 to start paused.