顺带一提，擅长”打脸直觉”的系列我又补了一篇：Python 字符串与正则性能深度实战：7 组实测数据告诉你正则不总是更快。里面用 timeit 实测了 7 组场景，好些结论跟我原本以为的正好相反，强烈建议一读。
凌晨三点，你的脚本吃了 8GB 内存
事情是这样的。那天我写了个脚本处理 1000 万行访问日志，提取某些 pattern、聚合、排序、取 Top 100。思路很简单：读文件 → 塞进 list → filter → map → sort → 切片。
跑起来之后，htop 显示内存从 200MB 一路飙升到 8.2GB，然后 OOM Killer 把它杀了。我盯着那个 results = [] 愣了几秒——每一行中间结果都在 list 里活着，1000 万行，每行就算只占 200 字节，那也是 2GB。加上 Python 对象开销、字符串 intern、list 的预留空间……8GB 真不多。
你猜怎么着？我花了 20 分钟把整套逻辑重写成 itertools 惰性管道，内存占用从 8GB 降到 180MB，速度反而快了 2.7 倍。
这不是魔法。这是 Python 标准库里最被低估的模块——itertools。
itertools 到底解决了什么问题？
一句话：它让你用「管道」思维写数据处理，而不是用「容器」思维。
传统的 for 循环写法是声明式地创建中间结果：
# 容器思维：每一步都生成一个完整 list
lines = [line.strip() for line in f] # 1000 万元素的 list
errors = [l for l in lines if "ERROR" in l] # 又 200 万
parsed = [parse(l) for l in errors] # 又 200 万
sorted_parsed = sorted(parsed, key=lambda x: x.timestamp)
top_100 = sorted_parsed[:100]
每一步都在堆上分配一个新 list。内存 = 5 个 list 的总和。如果你用 sys.getsizeof 算一下，一个 1000 万元素的 list 光是对象头就占 ~80MB，加上引用指针 ~80MB，加上每个字符串对象本身……
itertools 的惰性管道思维是这样的：
# 管道思维：没有中间 list，数据像水流一样穿过每一步
from itertools import islice
pipeline = (
parse(line.strip()) # 1. 解析
for line in f
if "ERROR" in line # 2. 过滤
)
# 到这为止，没有处理任何数据。f 还没开始读。
# 3. 排序、取 Top 100
sorted_errors = sorted(pipeline, key=lambda x: x.timestamp)
top_100 = sorted_errors[:100]
生成器表达式本身就是 itertools 的好朋友。配合 itertools 的函数，你可以搭出极其复杂的管道而几乎不分配额外内存。
十个用得上的 itertools 函数，附真实场景
我不会把文档搬过来。下面是十个我在生产环境真正用过的，每个都有场景和代码。
1. islice — 惰性切片，不创建新 list
场景：你要跳过 CSV 文件的前 5 行注释，然后取第 6 到第 100000 行处理。用 list 切片会先读 100000 行到内存。
from itertools import islice
with open("huge_log.csv") as f:
# 跳过注释行 + 表头（前5行）
# 然后惰性取前 10 万行——不会一次性读到内存
data_rows = islice(f, 5, 100000)
for row in data_rows:
process(row)
islice(it, start, stop, step) 跟 list 切片语法一模一样，但它返回迭代器，不是 list。适用于任何迭代器的切片操作，包括无限流。
2. groupby — 分组聚合不靠 pandas
场景：日志按小时分组统计 ERROR 计数。
from itertools import groupby
from datetime import datetime
def hour_key(line):
ts = line.split()[0] # "2026-08-06T14:32:01"
return ts[:13] # "2026-08-06T14"
# ⚠️ groupby 要求输入已按键排序！未排序会导致分组错误
lines = sorted(open("app.log"), key=hour_key)
for hour, group in groupby(lines, key=hour_key):
count = sum(1 for _ in group)
print(f"{hour} | {'#' * (count // 10)} {count}")
重要但是注意：
groupby跟 SQL 的GROUP BY不一样——它只合并相邻的相同 key。不排序就直接用它，你会得到分散的小组而不是一个完整组。这是新手踩坑率最高的 itertools 函数。
3. chain / chain.from_iterable — 扁平化嵌套结构
场景：多个日志文件的列表，要逐个遍历每一行。不用写嵌套循环，不用 extend 到一个大 list。
from itertools import chain
log_files = ["app1.log", "app2.log", "app3.log"]
# ❌ 容器思维：把所有行读进一个巨型 list
all_lines = []
for fname in log_files:
all_lines.extend(open(fname).readlines())
# ✅ 管道思维：惰性遍历，内存里始终只有一个文件的内容
for line in chain.from_iterable(open(f) for f in log_files):
if "timeout" in line:
alert(line)
chain.from_iterable 比 chain(*iterables) 好在它接受单个可迭代对象，不会先展开参数——拿一个生成器丢进去就行。处理嵌套结构时这是你的 go-to。
4. accumulate — 运行累计，比如算库存水位
场景：每天有进货和出货量，要算每日的库存水位曲线。用 for 循环要维护一个 running sum 变量。
from itertools import accumulate
daily_deltas = [50, -20, 30, -100, 70, -10] # 每日净变化
inventory = list(accumulate(daily_deltas, initial=100))
# [100, 150, 130, 160, 60, 130, 120]
你也可以传自定义函数——比如算最大回撤：
prices = [100, 105, 98, 102, 110, 95]
running_max = accumulate(prices, max)
# [100, 105, 105, 105, 110, 110]
drawdown = [p / m - 1 for p, m in zip(prices, running_max)]
# [0.0, 0.0, -0.067, -0.029, 0.0, -0.136]
5. dropwhile / takewhile — 按条件截断流
场景：跳过日志开头的启动信息，从第一条业务日志开始处理。用 flag 变量很丑陋。
from itertools import dropwhile
def is_startup(line):
return "STARTUP" in line or "INIT" in line or "Loading" in line
# 跳过所有启动行，从第一条业务行开始
business_lines = dropwhile(is_startup, open("app.log"))
for line in business_lines:
process(line)
takewhile 相反：满足条件就继续，不满足就断流。
from itertools import takewhile
# 只处理时间在 2026-08-06 的行（假设日志按时间排序）
for line in takewhile(lambda l: l.startswith("2026-08-06"), open("app.log")):
process(line)
6. pairwise — 滑动窗口，不用索引
场景：计算时序数据中相邻两点的变化率。老写法：for i in range(len(data) - 1)。
from itertools import pairwise
timestamps = [1.0, 1.05, 1.12, 1.08, 1.15]
for prev, curr in pairwise(timestamps):
change = (curr - prev) / prev * 100
print(f"{prev:.2f} → {curr:.2f}: {change:+.2f}%")
Python 3.10 加入。以前要用 zip(data, data[1:])——但 data[1:] 会拷贝 list。pairwise 是纯惰性的。
7. filterfalse — 反向过滤取异常数据
场景：从正常数据里取出异常值。filter 是取通过的，filterfalse 取被筛掉的。
from itertools import filterfalse
def is_valid(record):
return record.status == 200 and record.latency_ms < 1000
# 正常数据用 filter
ok_records = filter(is_valid, all_records)
# 异常数据用 filterfalse——不用写 `lambda r: not is_valid(r)`
bad_records = filterfalse(is_valid, all_records)
8. tee — 多路复用同一个迭代器
场景：你要同时算总和和平均值，但不能遍历两次（迭代器只能消费一次）。
from itertools import tee
import statistics
it1, it2 = tee(read_huge_dataset(), 2)
total = sum(it1)
avg = statistics.mean(it2) # it2 独立消费
print(f"Sum: {total}, Avg: {avg:.2f}")
⚠️ 但是注意：
tee内部用队列缓存数据。如果两个消费者速度差异很大（比如一个在 sort 另一个在 filter），队列会涨到跟原迭代器等大——等于把惰性管道的优势白给了。两个消费者速度差不多时才用它。
9. zip_longest — 不等长迭代器对齐
场景：两个数据源长度不一致，但你希望全部对齐，缺失的用默认值填充。
from itertools import zip_longest
a_series = [100, 105, 102]
b_series = [200, 201]
# zip 会丢掉 102
list(zip(a_series, b_series)) # [(100, 200), (105, 201)]
# zip_longest 用 fillvalue 补齐
list(zip_longest(a_series, b_series, fillvalue=None))
# [(100, 200), (105, 201), (102, None)]
10. product / combinations — 不是玩具，生产环境也在用
场景：A/B 测试中生成所有参数组合，不用写四层嵌套循环。
from itertools import product
# 参数网格搜索
batch_sizes = [32, 64, 128]
learning_rates = [1e-3, 1e-4, 1e-5]
optimizers = ["adam", "sgd"]
for bs, lr, opt in product(batch_sizes, learning_rates, optimizers):
run_experiment(batch_size=bs, learning_rate=lr, optimizer=opt)
# 3 × 3 × 2 = 18 次实验，没有 3 层嵌套
combinations 也实用：比如从 50 个候选特征中选 3 个做特征组合评估。
实测：手动循环 vs itertools 管道
我写了一段对比代码，处理一个 500 万行的模拟日志文件（~600MB）：
- 过滤出 ERROR 行
- 按小时分组统计
- 取 Top 20 的错误密度时段
手动循环版（容器思维）：
def manual_loop(filename):
with open(filename) as f:
lines = f.readlines() # <- 600MB 全吃进内存
errors = []
for line in lines:
if "ERROR" in line:
errors.append(line)
by_hour = {}
for line in errors:
hour = line.split()[0][:13]
by_hour.setdefault(hour, []).append(line)
result = []
for hour, errs in by_hour.items():
result.append((len(errs), hour))
result.sort(reverse=True)
return result[:20]
itertools 管道版：
from itertools import islice, groupby
def itertools_pipeline(filename):
def hour_key(line):
return line.split()[0][:13]
with open(filename) as f:
errors = (line for line in f if "ERROR" in line)
sorted_errors = sorted(errors, key=hour_key)
groups = groupby(sorted_errors, key=hour_key)
counts = ((sum(1 for _ in g), h) for h, g in groups)
sorted_counts = sorted(counts, reverse=True)
return list(islice(sorted_counts, 20))
| 指标 | 手动循环 | itertools 管道 | 差距 |
|---|---|---|---|
| 峰值内存 | 2.1 GB | 0.18 GB | 11.7× |
| 总耗时 | 12.3s | 4.6s | 2.7× |
| 代码行数 | 17 行 | 8 行（核心） | 2.1× |
根本原因不是”itertools 更快”，而是手动版在 f.readlines() 和中间 list 上花掉了大量内存分配和 GC 时间。itertools 版的 sorted() 调用是唯一的内存峰值点，但 sorted 内部用 Timsort 的临时数组，远小于全量 list 的 Python 对象开销。
组合出真正的数据管道：一个完整例子
这是我真实用过的日志分析脚本的简化版：
from itertools import (
islice, chain, groupby, filterfalse, dropwhile, pairwise
)
from collections import Counter
from datetime import datetime
def analyze_error_bursts(*log_files):
# 分析多个日志文件中的错误突发
# 输出：按时间排序的 Top 10 错误突发区间
# Step 1: 多文件惰性遍历，跳过启动行
all_lines = chain.from_iterable(open(f) for f in log_files)
business_lines = dropwhile(lambda l: "STARTUP" in l, all_lines)
# Step 2: 只取 ERROR 行 + 解析时间戳
errors = (
(line.split()[0], line)
for line in business_lines
if "ERROR" in line
)
# Step 3: 按 5 分钟窗口分组（pairwise 做滑动窗口）
sorted_errors = sorted(errors, key=lambda x: x[0])
windows = []
for (t1, _), (t2, _) in pairwise(sorted_errors):
windows.append(get_5min_window(t1, t2))
# Step 4: 统计每个窗口的错误类型分布
window_counts = Counter(windows)
return window_counts.most_common(10)
# 运行
results = analyze_error_bursts(
"/var/log/app1.log",
"/var/log/app2.log",
"/var/log/app3.log"
)
for window, count in results:
print(f"{window}: {count} errors")
这个脚本的内存峰值只取决于 sorted 这一步——因为必须要排序才能用 groupby。如果数据本身已经按时间写入（日志通常如此），连 sorted 这一步都可以省掉，那就是真正的 O(1) 内存、单次遍历。
两个常见踩坑
坑1：groupby 不排序就用
groupby 跟 SQL GROUP BY 完全不同。它只合并相邻的相同 key。
data = [("A", 1), ("B", 2), ("A", 3), ("B", 4)]
# ❌ 错误：不排序直接 groupby
for k, g in groupby(data, key=lambda x: x[0]):
print(k, list(g))
# A [("A", 1)]
# B [("B", 2)]
# A [("A", 3)] <- A 被拆成两组！
# B [("B", 4)]
# ✅ 正确：先排序
data_sorted = sorted(data, key=lambda x: x[0])
for k, g in groupby(data_sorted, key=lambda x: x[0]):
print(k, list(g))
# A [("A", 1), ("A", 3)]
# B [("B", 2), ("B", 4)]
坑2：tee 的两个消费者速度差太大
tee 用内部 deque 缓存数据。如果一个消费者跑得很快（比如做 sum），另一个跑得很慢（比如做 sorted），deque 会撑到跟原迭代器等大——等于把所有数据都缓存了。两个消费者的速度差异超过 2-3 倍就不要用 tee——要么重新生成迭代器，要么用磁盘临时存储。
FAQ
Q: itertools 是不是只适合小数据？大数据还是得用 Spark？
反了。itertools 在数据量超过内存时优势最大——因为它是惰性计算，不分配中间 list。单机能处理的数据量上限取决于你的耐心和磁盘 I/O，而不是内存。我个人在 16GB 笔记本上用它处理过 200GB 的日志文件（逐行读取，itertools 管道，结果写入 SQLite）。超过 500GB 才考虑 Spark。但你得接受一个现实：比 Spark 写起来舒服，比 pandas 内存友好——但调试起来痛苦，因为数据不可见。
Q: 生成器表达式和 itertools 怎么选？
简单过滤/映射用生成器表达式就够了：(x*2 for x in data if x > 0)。需要分组（groupby）、累积（accumulate）、扁平化（chain）、多路复用（tee）时用 itertools。两者混用效果最好——生成器表达式做单步操作，itertools 做结构化操作。
Q: itertools 会不会比手动循环慢？毕竟是函数调用？
itertools 的函数是用 C 实现的（在 CPython 的 _itertools 模块里），函数调用开销可以忽略。真正的性能瓶颈是内存分配——你手动循环时创建的中间 list。itertools 不分配这些 list，所以整体反而更快。我上面的实测数据就是这个结论。
Q: 怎么调试惰性管道？数据不可见啊。
两个技巧：一是在管道中间插入 islice(pipe, 10) 看一眼前 10 条数据；二是用 tee 分出一路打印日志。但通常我的做法是在 ipython 里把生成器表达式 list() 化，确认逻辑对之后再塞回管道。
总结
itertools 的核心思想就一句话：别攒数据，让数据流过去。
你不需要记全部函数。会用这 6 个就能处理 80% 的数据管道场景：islice、chain、groupby、accumulate、dropwhile、pairwise。剩下的遇到再查文档。
下次你写 results = [] 然后 for item in data: results.append(transform(item)) 的时候，停下来想一想：我真的需要一个 list 吗？
通常不需要。
如果你是 IT 同行而且遇到过 Python 内存问题，我之前还写过Python 内存泄漏排查、memoryview 零拷贝、__slots__ 内存优化——这几篇和本文是同一个方向的系列，感兴趣的可以连着看。
如果对生成器底层的 yield、send、throw、close 机制还不熟，建议先看这篇Python 生成器深度实战。