Home
Explore Uedu
Student Console
Register as Member/Login
(2) In future presentations of the research findings, in addition to the course project website and public presentations, your real name and personal information will not appear in this research report. If you are interested in the research results, we can provide you with an executive summary after the study is completed.
問卷中心
Teacher Console
Course Setup
Support & Messages
Uptime Data

UeduGPTs

--

Jupyters

0

Local AI

--

CISOSE26 本地 AI UG26
政治大學 AQI 18 29°C

AI Reply Desktop Notifications

Show a desktop notification when the AI TA finishes replying

Chat Message Notifications

Notify me when classmates post messages in the forum

Sound notification

Play an alert sound whenever there is a new notification

常用標準庫與生態

標準庫的進階心法:惰性迭代、函數式工具與高效能容器

從「有哪些工具」進階到「用對的方式思考」——itertools、functools、pathlib、dataclasses 與 heapq 如何讓你不爆記憶體地處理巨量資料

一個惰性的問題:如何不爆記憶體地處理十億筆資料?

你已經會用 Counterjsondatetime,也懂得開虛擬環境。現在來個進階情境:助教給你一個 8 GB 的日誌檔,要你算出「每個 IP 出現幾次,並找出 PM2.5 連續超標最久的時段」。你的筆電只有 8 GB 記憶體,open(f).read() 一行就會把程式打掛。

入門篇談的是「標準庫有哪些工具」;這篇要談的是「標準庫如何讓你用對的方式思考」——惰性求值(lazy evaluation)、函數式組合(functional composition)、以及一批被低估的高效能容器。這些不是炫技,而是讓你寫出能跑得動、又讀得懂的程式的關鍵。整箱「電池」裡,這幾顆才是專業工程師每天在用的。

常用標準庫與生態進階概念示意圖

itertools:把迴圈變成可組合的零件

入門時我們寫迴圈用 for x in lst,一次把整個串列載進記憶體。itertools 提供的是迭代器(iterator):它不預先算出所有結果,而是「被要求一個才生一個」。這就是惰性求值,也是處理巨量資料的核心。

先看幾個常用零件:

import itertools as it

# count:無限計數器(永遠不會停,靠外部條件中斷)
for i in it.count(10, 2):       # 從 10 開始,每次 +2
    if i > 16:
        break
    print(i, end=' ')           # 輸出:10 12 14 16

print()

# accumulate:累積運算(預設累加,可傳入自訂函式)
print(list(it.accumulate([1, 2, 3, 4])))        # 輸出:[1, 3, 6, 10]
print(list(it.accumulate([3, 1, 4, 1, 5], max))) # 輸出:[3, 3, 4, 4, 5](跑動最大值)

# groupby:把「相鄰且相同」的元素分組(注意:只看相鄰!)
data = [('A', 1), ('A', 2), ('B', 3), ('A', 4)]
for key, group in it.groupby(data, key=lambda t: t[0]):
    print(key, [t[1] for t in group])
# 輸出:
# A [1, 2]
# B [3]
# A [4]   ← 最後的 A 因為不相鄰,自成一組

最後那個 groupby 是經典陷阱:它只合併相鄰的相同鍵,所以用之前通常要先 sorted()。這跟 SQL 的 GROUP BY 行為不同,很多人在這裡踩雷。

itertools 真正的威力在組合。假設要對一份成績做「滑動視窗」分析——每三筆算一次平均:

import itertools as it

def sliding_window(iterable, n):
    """產生長度為 n 的滑動視窗,全程惰性、不複製整個序列。"""
    it1 = iter(iterable)
    window = tuple(it.islice(it1, n))   # 先取前 n 個
    if len(window) == n:
        yield window
    for x in it1:
        window = window[1:] + (x,)      # 滑動:丟掉最左、補上最右
        yield window

scores = [60, 70, 80, 90, 100]
for w in sliding_window(scores, 3):
    print(w, '平均', sum(w) / len(w))
# 輸出:
# (60, 70, 80) 平均 70.0
# (70, 80, 90) 平均 80.0
# (80, 90, 100) 平均 90.0

整段程式從頭到尾沒有把整份資料複製一份,無論 scores 是 5 筆還是 5 億筆,記憶體用量幾乎不變。這就是惰性求值的承諾。Python 3.10 起標準庫直接內建了 itertools.pairwise,3.12 起更有 itertools.batched,你不用每次都自己手寫。

functools:替函數加上記憶與快取

functools 是「對函數做手術」的工具箱。三個你一定會用到:reducepartiallru_cache

lru_cache 是其中最實用的——它替函數加上一層自動快取(memoization),把同樣參數的重複計算直接省掉。經典例子是費氏數列:

from functools import lru_cache

@lru_cache(maxsize=None)
def fib(n):
    if n < 2:
        return n
    return fib(n - 1) + fib(n - 2)

print(fib(100))          # 輸出:354224848179261915075(瞬間完成)
print(fib.cache_info())  # 輸出:CacheInfo(hits=98, misses=101, ...)

沒有快取的樸素遞迴 fib,時間複雜度是指數級的 $O(\phi^n)$,算 fib(40) 就要等好幾秒;加上 @lru_cache 後每個 n 只算一次,複雜度直接降到 $O(n)$。cache_info() 告訴你快取命中了 98 次——這 98 次都是省下來的重複運算。

partial 則是「先把部分參數填好,做出一個新函數」:

from functools import partial

def power(base, exp):
    return base ** exp

square = partial(power, exp=2)   # 把 exp 固定成 2
cube = partial(power, exp=3)
print(square(5), cube(5))        # 輸出:25 125

這在需要傳「回呼函數(callback)」給別的 API(例如 sortedkey、GUI 的事件處理器)時特別好用,不必為了固定一個參數而寫一堆 lambda

pathlib:用物件取代字串拼路徑

入門篇用 os.path.join 組路徑。現代 Python(3.4 起)更推薦 pathlib,它把路徑當成物件,用運算子 / 來串接,可讀性大幅提升:

from pathlib import Path

root = Path('data') / 'logs' / '2026'   # 用 / 串路徑,跨平台自動處理分隔符
print(root)                              # 輸出:data/logs/2026(Windows 顯示反斜線)

p = Path('report.final.csv')
print(p.suffix)    # 輸出:.csv         (副檔名)
print(p.stem)      # 輸出:report.final (去掉最後副檔名)
print(p.name)      # 輸出:report.final.csv

# 找出某資料夾下所有 .py 檔(遞迴),全程惰性
for py in Path('.').rglob('*.py'):
    print(py)

# 讀寫檔案一行搞定,不必 open/close
Path('note.txt').write_text('你好', encoding='utf-8')
print(Path('note.txt').read_text(encoding='utf-8'))  # 輸出:你好

pathlib 不只是語法糖:Path 物件帶有 .exists().is_dir().parent.glob() 等方法,把「字串操作」與「檔案系統查詢」統一成一個介面。比起在 osos.pathglobshutil 之間跳來跳去,pathlib 讓檔案處理程式碼乾淨許多。新專案請優先用它。

dataclasses 與 enum:讓資料結構自我描述

當你要表示一筆「結構化的資料」(一個學生、一筆訂單、一個測站讀數),與其用容易打錯鍵名的 dict,不如用 dataclasses。它幫你自動生成 __init____repr____eq__,省掉一堆樣板程式碼:

from dataclasses import dataclass, field

@dataclass
class Reading:
    station: str
    pm25: float
    valid: bool = True               # 有預設值的欄位
    tags: list = field(default_factory=list)  # 可變預設值要用 field

r1 = Reading('中央站', 35.2)
r2 = Reading('中央站', 35.2)
print(r1)              # 輸出:Reading(station='中央站', pm25=35.2, valid=True, tags=[])
print(r1 == r2)        # 輸出:True(自動比對所有欄位,dict 做不到這麼簡潔)

注意 tags 那行:絕對不要寫 tags: list = []。Python 的預設值只在定義函式時計算一次,所有實例會共用同一個串列——這是著名的「可變預設參數」陷阱。dataclasses 強制你用 field(default_factory=list),每個實例各拿一個新串列,從語言層面幫你避開這個坑。

搭配 enum 可以讓「有限選項」變成不可打錯的常數:

from enum import Enum

class Level(Enum):
    GOOD = 1
    MODERATE = 2
    UNHEALTHY = 3

def classify(pm25):
    if pm25 <= 15:
        return Level.GOOD
    elif pm25 <= 35:
        return Level.MODERATE
    return Level.UNHEALTHY

print(classify(40))         # 輸出:Level.UNHEALTHY
print(classify(40).name)    # 輸出:UNHEALTHY

比起到處散落字串 'unhealthy'(打錯一個字母就變成靜默的 bug),enum 讓非法值在寫程式時就無法存在。

高效能容器:deque 與 heapq

入門篇講了 Counterdefaultdictcollections 還藏著一個 deque(雙端佇列),以及隔壁的 heapq(堆積)——這兩個是處理「排程」「串流」問題的利器。

deque 的兩端進出都是 $O(1)$,而普通 list 在開頭 insert(0, x)pop(0) 是 $O(n)$(要搬移後面所有元素)。需要「先進先出佇列」或「固定長度的滑動緩衝」時,用 deque

from collections import deque

# maxlen 讓它變成自動丟棄舊資料的固定視窗——最近 3 筆而已
recent = deque(maxlen=3)
for x in [10, 20, 30, 40, 50]:
    recent.append(x)
    print(list(recent))
# 輸出:
# [10]
# [10, 20]
# [10, 20, 30]
# [20, 30, 40]   ← 10 被自動擠出去
# [30, 40, 50]

heapq 把普通 list 當成最小堆積(min-heap)操作,讓你以 $O(\log n)$ 取出最小值。要從十億筆資料裡找「最大的 10 筆」而不排序整份資料時,它是標準答案:

import heapq

data = [5, 1, 8, 3, 9, 2, 7]
print(heapq.nlargest(3, data))   # 輸出:[9, 8, 7]
print(heapq.nsmallest(2, data))  # 輸出:[1, 2]

# 當作優先佇列:每次 heappop 都拿到目前最小的
h = []
heapq.heappush(h, (3, '低優先'))
heapq.heappush(h, (1, '高優先'))
heapq.heappush(h, (2, '中優先'))
print(heapq.heappop(h))          # 輸出:(1, '高優先')

heapq.nlargest(10, huge_stream) 只需要維護一個 10 個元素的堆積,記憶體 $O(k)$ 而非 $O(n)$,完美呼應開頭那個「8 GB 檔案、8 GB 記憶體」的限制。

看一個例子:串流式日誌分析

現在把惰性迭代、Counterdeque 組起來,回答開頭的問題——一行也不把整個檔案讀進記憶體,逐行統計 IP 次數,同時追蹤 PM2.5 連續超標最久的時段:

from collections import Counter

def analyze_log(path, threshold=35):
    ip_counter = Counter()
    cur_streak = 0      # 目前連續超標筆數
    best_streak = 0     # 史上最長連續超標

    with open(path, encoding='utf-8') as f:
        for line in f:                  # 逐行讀,記憶體只放一行
            parts = line.split(',')
            if len(parts) < 2:
                continue
            ip, pm25 = parts[0], float(parts[1])
            ip_counter[ip] += 1

            if pm25 > threshold:
                cur_streak += 1
                best_streak = max(best_streak, cur_streak)
            else:
                cur_streak = 0          # 一旦低於門檻就歸零

    return ip_counter.most_common(3), best_streak

# 假設 log.csv 每行格式為「IP,PM2.5」
# top3, longest = analyze_log('log.csv')
# print('最常出現的 IP:', top3)
# print('最長連續超標筆數:', longest)

關鍵在 for line in f:檔案物件本身就是迭代器,Python 一次只把一行載進記憶體。無論 log.csv 是 8 MB 還是 8 GB,這段程式的記憶體用量都維持在「一行字串 + 一個 Counter」的等級。這就是用標準庫的惰性機制,把「資料量」和「記憶體量」解耦的實作典範。

contextlib:自己做 with 區塊

入門時你用過 with open(...),它保證檔案無論如何都會被關閉。這個「資源自動清理」的機制叫上下文管理器(context manager)。contextlib.contextmanager 讓你用一個 yield 就做出自己的 with

import time
from contextlib import contextmanager

@contextmanager
def timer(label):
    start = time.perf_counter()
    try:
        yield                        # 這裡把控制權交給 with 區塊內的程式
    finally:
        elapsed = time.perf_counter() - start
        print(f'{label} 花了 {elapsed:.4f} 秒')

with timer('排序一百萬筆'):
    sorted(range(1_000_000, 0, -1))
# 輸出:排序一百萬筆 花了 0.0xxx 秒

yield 之前是「進入時做的事」,finally 裡是「離開時保證做的事」——即使區塊內拋出例外,計時與清理照樣執行。理解這個模式,你就能優雅地包裝「開關資料庫連線」「暫時切換工作目錄」「鎖定再解鎖」這類成對操作。

重點回顧

  • 惰性求值是處理巨量資料的核心itertools 與檔案物件都是迭代器,「被要求才生一個」,讓記憶體用量與資料總量脫鉤。
  • functools.lru_cache 一行把指數級遞迴降成線性:自動快取重複計算,搭配 partial 固定參數,是函數式工具箱的兩大支柱。
  • pathlib 取代字串拼路徑dataclasses 取代易錯的 dictenum 取代散落的魔術字串——讓資料結構自我描述、在寫程式時就擋掉錯誤。
  • 選對容器決定複雜度deque 兩端進出 $O(1)$、heapq 取極值 $O(\log n)$,比硬用 list 快上數量級。
  • contextlib 讓你自製 with,用 try/finally 保證資源清理,即使例外發生也不漏。

深入探討(研究所視角)

迭代器協定與惰性求值的代價。Python 的 for 迴圈底層是迭代器協定(iterator protocol):物件實作 __iter____next__,每次呼叫 next() 回傳下一個值,耗盡時拋 StopIteration。生成器(generator)是這個協定的語法糖——yield 會把函式編譯成一台暫停/恢復的狀態機,局部變數凍結在堆疊框架(frame object)裡。惰性求值的好處是空間複雜度從 $O(n)$ 降到 $O(1)$,但代價是「只能走一遍」且「無法隨機存取」:一個迭代器被消費完就空了,這正是初學者誤把 map 物件 print 兩次第二次變空的根因。理解 itertools.tee(複製迭代器但需緩衝)與「拉取式(pull-based)資料流」的本質,是讀懂 asyncio、串流處理框架(如 Apache Beam)乃至資料庫查詢執行引擎的基礎。

CPython 標準庫的 C 加速層。很多人以為標準庫只是「方便」,其實它常常還「快」。collections.dequeheapq 的關鍵迴圈、itertools 的多數函式、json 的解析器、re 的引擎,都以 C 實作(_collections_heapq_json 等擴充模組)。這意味著 heapq.nlargestCounter 的內層迴圈跑在 C 速度,繞過了 Python 直譯器的逐行開銷與全域直譯器鎖(GIL)在純 Python 迴圈上的瓶頸。因此一條經驗法則是:能用標準庫的內建函式表達的運算,幾乎一定比你自己寫 Python 迴圈快——這也是為何 sumsorteditertools.chain 往往勝過手刻版本。真正的效能調校第一步,常常不是改演算法,而是把熱點迴圈換成 C 加速的標準庫呼叫,或進一步交給 NumPy 的向量化。

標準庫作為「正確性的預設值」dataclasses 強制 default_factorysecrets 取代 random 做金鑰、decimal 取代 float 做金額、fractions 保留精確有理數——這些設計都把「容易出錯的決定」變成「標準庫替你做對的預設」。以 decimal.Decimal 為例,0.1 + 0.2 != 0.3 是 IEEE 754 二進位浮點的本質限制,金融與計量場景用 Decimal 可得精確十進位運算,代價是速度較慢。研究所層級的工程素養,很大一部分就是知道「在什麼情境該換掉預設工具」:高精度計算選 decimal/fractions、安全亂數選 secrets、跨程序平行選 multiprocessing 而非 threading(繞過 GIL)。標準庫不只是電池,它是一套經過數十年實戰淬鍊、把正確選擇編碼進 API 的集體智慧——讀它的原始碼(Lib/ 目錄)本身就是一門最好的軟體工程課。

AI 共讀助教正在陪你讀:標準庫的進階心法:惰性迭代、函數式工具與高效能容器
嗨!我是這篇文章的共讀助教,只根據〈標準庫的進階心法:惰性迭代、函數式工具與高效能容器〉的內容回答。可以問我「解釋某段」「舉個例子」「出題考我」,或反白文中段落後點下方「解釋選取段落」。