Home
探索 Uedu
學生控制台
註冊會員/登入
研究知情同意中心
教師控制台
課程設定
支援與訊息
Uptime 數據

UeduGPTs

--

Jupyters

4

UG26 CISOSE26
臺北 AQI 46 · 臺中 AQI 28 · 臺南 AQI 24 · 高雄 AQI 33

AI 回覆桌面通知

AI 助教回覆完成時顯示桌面通知

聊天訊息通知

同學在討論區發送訊息時通知

聲音通知

每當有新通知時播放提示音

常用標準庫與生態

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

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