Python itertools 高级数据流水线：从组合数学到高效迭代

1
# 从 API 分页取数据，过滤无效项，分组后处理
2
data = []
3
for page in range(100):
4
    items = fetch_api(page)
5
    for item in items:
6
        if item.get("status") == "active" and item.get("value") > 0:
7
            data.append(item)
8

9
# 然后按类型分组
10
groups = {}
11
for item in data:
12
    key = item["type"]
13
    if key not in groups:
14
        groups[key] = []
15
    groups[key].append(item)

1
"""
2
log_analyzer.py - 使用 itertools 构建的日志分析管道
3
分析 Nginx 访问日志，提取关键指标
4
"""
5
from itertools import chain, islice, filterfalse, groupby, accumulate
6
from collections import Counter
7
import re
8
from datetime import datetime
9

10
# 模拟日志行
11
SAMPLE_LOGS = [
12
    '192.168.1.1 - - [27/May/2026:10:00:01 +0800] "GET /api/users HTTP/1.1" 200 1234 0.045',
13
    '192.168.1.2 - - [27/May/2026:10:00:02 +0800] "POST /api/login HTTP/1.1" 401 89 0.123',
14
    '10.0.0.1 - - [27/May/2026:10:00:03 +0800] "GET /static/app.js HTTP/1.1" 200 45678 0.002',
15
    '192.168.1.1 - - [27/May/2026:10:00:04 +0800] "GET /api/orders HTTP/1.1" 500 234 1.456',
16
    '10.0.0.2 - - [27/May/2026:10:00:05 +0800] "GET /health HTTP/1.1" 200 2 0.001',
17
    '',  # 空行
18
    '192.168.1.3 - - [27/May/2026:10:00:06 +0800] "GET /api/users/42 HTTP/1.1" 200 567 0.034',
19
    '# maintenance log',  # 注释行
20
    '10.0.0.1 - - [27/May/2026:10:00:07 +0800] "GET /api/products HTTP/1.1" 200 8901 0.078',
21
    '192.168.1.1 - - [27/May/2026:10:00:08 +0800] "DELETE /api/users/99 HTTP/1.1" 403 56 0.012',
22
]
23

24
LOG_PATTERN = re.compile(
25
    r'(\S+)\s+\S+\s+\S+\s+\[([^\]]+)\]\s+"(\S+)\s+(\S+)\s+\S+"\s+(\d+)\s+(\d+)\s+(\S+)'
26
)
27

28
def parse_log_line(line: str) -> dict | None:
29
    """解析单条日志行"""
30
    m = LOG_PATTERN.match(line)
31
    if not m:
32
        return None
33
    return {
34
        "ip": m.group(1),
35
        "time": m.group(2),
36
        "method": m.group(3),
37
        "path": m.group(4),
38
        "status": int(m.group(5)),
39
        "size": int(m.group(6)),
40
        "latency": float(m.group(7)),
41
    }
42

43
def analyze_logs(log_lines):
44
    """分析日志流水线"""
45
    # 阶段1：过滤空行和注释
46
    clean_lines = filterfalse(
47
        lambda l: not l.strip() or l.startswith("#"),
48
        log_lines,
49
    )
50

51
    # 阶段2：解析
52
    parsed = (parse_log_line(line) for line in clean_lines)
53
    parsed = filterfalse(lambda x: x is None, parsed)
54
    parsed = list(parsed)  # 小数据集物化，大数据集保持迭代器
55

56
    if not parsed:
57
        print("没有有效的日志记录")
58
        return
59

60
    # 阶段3：统计
61
    total_requests = len(parsed)
62

63
    # 状态码分布
64
    status_counter = Counter(entry["status"] for entry in parsed)
65

66
    # 按 IP 分组统计请求数（先排序再分组）
67
    sorted_by_ip = sorted(parsed, key=lambda x: x["ip"])
68
    ip_stats = {}
69
    for ip, group in groupby(sorted_by_ip, key=lambda x: x["ip"]):
70
        entries = list(group)
71
        ip_stats[ip] = {
72
            "count": len(entries),
73
            "avg_latency": sum(e["latency"] for e in entries) / len(entries),
74
            "errors": sum(1 for e in entries if e["status"] >= 400),
75
        }
76

77
    # 按路径分组
78
    path_counter = Counter(entry["path"] for entry in parsed)
79

80
    # 错误请求
81
    errors = [e for e in parsed if e["status"] >= 400]
82

83
    # 延迟百分位数（用累积方法近似）
84
    latencies = sorted(e["latency"] for e in parsed)
85
    p50_idx = int(len(latencies) * 0.5)
86
    p95_idx = int(len(latencies) * 0.95)
87
    p99_idx = int(len(latencies) * 0.99)
88

89
    # 输出报告
90
    print(f"\n{'='*60}")
91
    print(f"  日志分析报告")
92
    print(f"{'='*60}")
93
    print(f"  总请求数: {total_requests}")
94
    print(f"  错误率: {len(errors)/total_requests*100:.1f}%")
95
    print(f"\n  状态码分布:")
96
    for status, count in sorted(status_counter.items()):
97
        bar = "█" * count
98
        print(f"    {status}: {count:3d} {bar}")
99

100
    print(f"\n  Top 请求路径:")
101
    for path, count in path_counter.most_common(5):
102
        print(f"    {path:30s} {count}次")
103

104
    print(f"\n  IP 统计:")
105
    for ip, stats in sorted(ip_stats.items(), key=lambda x: -x[1]["count"]):
106
        print(f"    {ip}: {stats['count']}次请求, "
107
              f"平均延迟 {stats['avg_latency']*1000:.1f}ms, "
108
              f"错误 {stats['errors']}次")
109

110
    print(f"\n  延迟分布:")
111
    print(f"    P50: {latencies[p50_idx]*1000:.1f}ms")
112
    print(f"    P95: {latencies[p95_idx]*1000:.1f}ms")
113
    print(f"    P99: {latencies[p99_idx]*1000:.1f}ms")
114

115
# 运行分析
116
analyze_logs(SAMPLE_LOGS)

1
============================================================
2
  日志分析报告
3
============================================================
4
  总请求数: 8
5
  错误率: 37.5%
6

7
  状态码分布:
8
    200:   5 █████
9
    401:   1 █
10
    403:   1 █
11
    500:   1 █
12

13
  Top 请求路径:
14
    /api/users                     1次
15
    /api/login                     1次
16
    /static/app.js                 1次
17
    /api/orders                    1次
18
    /health                        1次
19

20
  IP 统计:
21
    192.168.1.1: 3次请求, 平均延迟 504.3ms, 错误 2次
22
    10.0.0.1: 2次请求, 平均延迟 40.0ms, 错误 0次
23
    ...
24

25
  延迟分布:
26
    P50: 34.0ms
27
    P95: 1456.0ms
28
    P99: 1456.0ms

1
import time
2
import tracemalloc
3
from itertools import islice, chain, filterfalse
4

5
# 模拟 100 万条数据
6
def generate_data(n):
7
    for i in range(n):
8
        yield {"id": i, "value": i * 2 + 1, "active": i % 3 != 0}
9

10
N = 1_000_000
11

12
# 方法1：传统列表处理
13
tracemalloc.start()
14
t1 = time.perf_counter()
15

16
data = list(generate_data(N))
17
filtered = [d for d in data if d["active"]]
18
transformed = [{"id": d["id"], "doubled": d["value"] * 2} for d in filtered]
19
result1 = list(islice(transformed, 100))
20

21
t1_end = time.perf_counter()
22
mem1 = tracemalloc.get_traced_memory()[1]
23
tracemalloc.stop()
24

25
# 方法2：itertools 流水线
26
tracemalloc.start()
27
t2 = time.perf_counter()
28

29
data_iter = generate_data(N)
30
pipeline = (
31
    {"id": d["id"], "doubled": d["value"] * 2}
32
    for d in data_iter
33
    if d["active"]
34
)
35
result2 = list(islice(pipeline, 100))
36

37
t2_end = time.perf_counter()
38
mem2 = tracemalloc.get_traced_memory()[1]
39
tracemalloc.stop()
40

41
print(f"传统列表:")
42
print(f"  时间: {t1_end - t1:.3f}s")
43
print(f"  内存峰值: {mem1 / 1024 / 1024:.1f} MB")
44
print(f"\nitertools 流水线:")
45
print(f"  时间: {t2_end - t2:.3f}s")
46
print(f"  内存峰值: {mem2 / 1024 / 1024:.1f} MB")
47

48
# 验证结果一致
49
assert result1 == result2, "结果不一致！"
50
print(f"\n结果验证: ✅ 一致")

1
from itertools import *
2
import collections
3

4
# 1. 分块（chunk）
5
def chunked(iterable, size):
6
    """将迭代器按固定大小分块"""
7
    it = iter(iterable)
8
    while True:
9
        chunk = list(islice(it, size))
10
        if not chunk:
11
            break
12
        yield chunk
13

14
# 2. 滑动窗口
15
def sliding_window(iterable, n):
16
    """大小为 n 的滑动窗口"""
17
    it = iter(iterable)
18
    window = collections.deque(islice(it, n), maxlen=n)
19
    if len(window) == n:
20
        yield tuple(window)
21
    for x in it:
22
        window.append(x)
23
        yield tuple(window)
24

25
# 3. 去重（保持顺序）
26
def unique(iterable):
27
    """惰性去重，保持首次出现的顺序"""
28
    seen = set()
29
    for item in iterable:
30
        if item not in seen:
31
            seen.add(item)
32
            yield item
33

34
# 4. 交错合并
35
def interleave(*iterables):
36
    """交错合并多个迭代器"""
37
    return chain.from_iterable(zip(*iterables))
38

39
# 5. 配对（相邻元素）
40
def pairwise(iterable):
41
    """(s0,s1), (s1,s2), (s2,s3), ..."""
42
    a, b = tee(iterable)
43
    next(b, None)
44
    return zip(a, b)
45

46
# 6. 限制迭代器的最大长度
47
def take(n, iterable):
48
    """取前 n 个元素"""
49
    return list(islice(iterable, n))
50

51
# 7. 消费整个迭代器（用于触发副作用）
52
def consume(iterator, n=None):
53
    """消费迭代器，可用于触发生成器副作用"""
54
    if n is None:
55
        collections.deque(iterator, maxlen=0)
56
    else:
57
        next(islice(iterator, n, n), None)