一. 前言:檔案很大,不代表要整包讀進 Python #
假設你手上有一個 8 GB 的 log。
你只想找出第一個 ERROR,或讀取已知 offset 附近的 64 bytes。
最直覺的寫法可能是:
from pathlib import Path
data = Path("app.log").read_bytes()
position = data.find(b"ERROR")
程式很短,代價卻很直接:先配置一大塊 Python 記憶體,再把整個檔案讀進來。
另一個方向是逐塊讀取。
它很適合串流處理,但你得自己處理 chunk 邊界、目前位置與跨區塊匹配。
標準庫 mmap 提供第三條路:把檔案的一段位址空間映射成像 bytearray 一樣的物件。
作業系統負責按需載入頁面,Python 則可以:
- 用索引或切片隨機讀取。
- 用
find()搜尋 bytes。 - 搭配正規表示式掃描內容。
- 在固定長度範圍內原地修改檔案。
- 只映射大型檔案的一個區段。
不過,
mmap不是「用了就一定比較快」的魔法。 它仍然會碰到 page fault、檔案描述符、offset 對齊與資源生命週期。 今天拍拍君就從安全的唯讀 mapping 開始,一路做到可重用的小工具。
二. 安裝:不用裝,但先認識 bytes 世界 #
mmap 是 Python 標準庫,不需要 pip install:
import mmap
本文範例適合在一般 CPython 桌面與伺服器環境執行。
mmap 不支援 WASI;Unix 與 Windows 的底層 constructor 也有差異。
跨平台程式最好優先使用共同介面:
mmap.mmap(
file_obj.fileno(),
length=0,
access=mmap.ACCESS_READ,
)
其中:
fileno()把 Python file object 的檔案描述符交給mmap。length=0代表映射整個現有檔案。ACCESS_READ明確宣告唯讀。 還有一個重要觀念:mapping 裡看到的是 bytes,不是自動解碼後的str。 所以搜尋字串時要寫:
needle = "拍拍君".encode("utf-8")
而不是直接把 Python 字串丟給 find()。
三. 第一個唯讀 mapping:像 bytes,也像 file #
先建立一個小檔案:
from pathlib import Path
path = Path("events.log")
path.write_bytes(
b"INFO boot\n"
b"INFO cache-ready\n"
b"ERROR database-timeout\n"
b"INFO shutdown\n"
)
接著把整個檔案映射成唯讀物件:
import mmap
with path.open("rb") as file_obj:
with mmap.mmap(
file_obj.fileno(),
length=0,
access=mmap.ACCESS_READ,
) as mm:
print(mm[:9])
print(mm.find(b"ERROR"))
print(mm.size())
輸出類似:
b'INFO boot'
27
64
mm[:9] 看起來像 bytes slicing。
mm.find() 也像 bytes 的搜尋方法。
但 mm 同時保有 file-like cursor:
with path.open("rb") as file_obj:
with mmap.mmap(
file_obj.fileno(),
0,
access=mmap.ACCESS_READ,
) as mm:
print(mm.readline())
print(mm.tell())
mm.seek(0)
print(mm.read(4))
這裡的索引與 cursor 是兩套相關但不同的操作方式。
mm[10:20] 不會移動 cursor;read()、readline() 與 seek() 會。
如果函式混用兩種風格,最好清楚註明 cursor 是否會改變。
四. 搜尋大型檔案:不要忘記「找不到」是 -1 #
把搜尋封裝成函式:
from pathlib import Path
import mmap
def find_first(path: Path, needle: bytes) -> int | None:
if not needle:
raise ValueError("needle 不可以是空 bytes")
if path.stat().st_size == 0:
return None
with path.open("rb") as file_obj:
with mmap.mmap(
file_obj.fileno(),
0,
access=mmap.ACCESS_READ,
) as mm:
position = mm.find(needle)
return None if position == -1 else position
使用方式:
position = find_first(Path("events.log"), b"database")
print(position)
拍拍君特別處理了兩件事。
第一,空檔案不能照常建立長度為 0 的 file mapping,所以先看 st_size。
第二,find() 找不到時回傳 -1。
如果直接把結果拿去切片:
position = mm.find(b"NOT-FOUND")
print(mm[position:position + 20])
你可能會從檔案尾端附近讀到內容,讓 bug 看起來像「搜尋成功但結果怪怪的」。
先把 -1 轉成 None,呼叫端會比較不容易誤用。
找出全部匹配位置 #
find() 可以指定起點,因此不必每次重頭掃描:
def find_all(mm: mmap.mmap, needle: bytes):
start = 0
while True:
position = mm.find(needle, start)
if position == -1:
return
yield position
start = position + len(needle)
使用:
with path.open("rb") as file_obj:
with mmap.mmap(
file_obj.fileno(),
0,
access=mmap.ACCESS_READ,
) as mm:
print(list(find_all(mm, b"INFO")))
這個版本回傳不重疊匹配。
若要允許重疊,例如在 b"aaaa" 裡找 b"aa",把下一個起點改成:
start = position + 1
五. 隨機存取:讀取已知 offset 附近的資料 #
mmap 很適合「已知位置,再取一小段」的工作。
例如建立簡單的 log preview:
def preview(
mm: mmap.mmap,
position: int,
radius: int = 24,
) -> bytes:
if position < 0 or position >= len(mm):
raise IndexError("position 超出 mapping 範圍")
start = max(0, position - radius)
stop = min(len(mm), position + radius)
return mm[start:stop]
使用:
with path.open("rb") as file_obj:
with mmap.mmap(
file_obj.fileno(),
0,
access=mmap.ACCESS_READ,
) as mm:
position = mm.find(b"ERROR")
if position != -1:
print(preview(mm, position).decode("utf-8"))
切片會建立新的 bytes 物件。 只取小範圍通常沒問題;如果一次切出數 GB,仍然會配置巨大記憶體。 也就是說,mapping 避免了「一開始就整包複製」,不代表後續操作永遠零複製。
六. 原地修改:ACCESS_WRITE 與固定長度規則 #
要讓修改寫回原檔,請用 r+b 開檔,並指定 ACCESS_WRITE:
status_path = Path("status.dat")
status_path.write_bytes(b"state=PENDING\n")
with status_path.open("r+b") as file_obj:
with mmap.mmap(
file_obj.fileno(),
0,
access=mmap.ACCESS_WRITE,
) as mm:
start = mm.find(b"PENDING")
if start == -1:
raise RuntimeError("找不到狀態欄位")
mm[start:start + 7] = b"SUCCESS"
mm.flush()
print(status_path.read_text())
輸出:
state=SUCCESS
關鍵是 PENDING 與 SUCCESS 都是 7 bytes。
對 mapping slice 賦值時,新資料長度必須等於原 slice 長度。
下面這樣不行:
mm[start:start + 7] = b"OK"
mmap 不是會自動把後面內容推開的文字編輯器。
如果欄位長度可能改變,比較安全的做法是寫到新檔,再用原子替換完成更新。
需要高階複製、搬移與替換流程時,可以接著看 Python shutil 實戰。
ACCESS_COPY:改得到 mapping,改不到原檔 #
如果只想做「假設修改」或臨時 patch,可以使用 copy-on-write:
with status_path.open("rb") as file_obj:
with mmap.mmap(
file_obj.fileno(),
0,
access=mmap.ACCESS_COPY,
) as mm:
mm[6:13] = b"PENDING"
print(mm[:])
print(status_path.read_bytes())
三種常用 access 模式可以這樣記:
| 模式 | mapping 可修改 | 寫回原檔 |
|---|---|---|
ACCESS_READ |
否 | 否 |
ACCESS_WRITE |
是 | 是 |
ACCESS_COPY |
是 | 否 |
把意圖寫成 access=,通常比直接操作平台特定的 flags 與 prot 更容易跨平台。 |
七. 只映射一段:offset 必須對齊 #
超大型檔案不一定要整份 mapping,但 offset 必須是 mmap.ALLOCATIONGRANULARITY 的倍數;在 Unix 上通常等於 page size。任意位置要先向下對齊,再用 delta 找回真正起點:
granularity = mmap.ALLOCATIONGRANULARITY
aligned = target_offset // granularity * granularity
delta = target_offset - aligned
with path.open("rb") as file_obj:
with mmap.mmap(
file_obj.fileno(),
delta + payload_size,
access=mmap.ACCESS_READ,
offset=aligned,
) as mm:
payload = mm[delta:delta + payload_size]
正式工具還要驗證範圍沒有越過 EOF,而且 mapping 期間檔案不會被其他程序截短。部分 mapping 很實用,但邊界條件更多;一般 log 搜尋先映射整份檔案會比較容易寫對。
八. 資源生命週期:先關 mapping,再關檔案 #
最推薦的結構是巢狀 with:
with path.open("rb") as file_obj:
with mmap.mmap(
file_obj.fileno(),
0,
access=mmap.ACCESS_READ,
) as mm:
consume(mm)
離開內層時先關 mapping,再離開外層關檔案。
這個順序清楚,而且例外發生時也會清理資源。
不要把 mm 傳出 with 後繼續使用:
def bad_open(path: Path) -> mmap.mmap:
with path.open("rb") as file_obj:
with mmap.mmap(
file_obj.fileno(),
0,
access=mmap.ACCESS_READ,
) as mm:
return mm
回傳時 mapping 已經被關閉,呼叫端拿到的是不能再讀的物件。
memoryview 也要先釋放 #
memoryview(mm) 可以避免某些中間複製,但它會匯出底層 buffer。如果 view 還活著就關閉 mapping,可能得到 BufferError;因此 view 的生命週期必須比 mapping 更短。
九. mmap、read、串流,怎麼選? #
沒有一個方式永遠贏。
| 需求 | 優先考慮 |
|---|---|
| 小檔案,整份內容都要用 | read() / Path.read_bytes() |
| 單向逐列轉換 | buffered file iteration |
| 大檔案內搜尋 bytes | mmap |
| 已知 offset 的隨機讀取 | mmap 或 seek() |
| 跨 S3、HTTP、本機的統一介面 | fsspec |
| 長度會改變的內容重寫 | 新檔寫入後替換 |
mmap 適合多次查看大型檔案的不同位置、bytes 搜尋與固定布局資料;它不適合遠端 object storage、只讀一次的簡單 ETL,或長度會大幅變化的文字重寫。
處理本機與遠端儲存抽象時,請看 Python fsspec 實戰。
只想把路徑組合、遍歷與副檔名處理寫乾淨,則 Python pathlib 教學 更直接。
十. 常見陷阱清單 #
mm[:]、大切片與bytes(mm)仍會建立副本,不是自動零複製。- mapping 是 bytes-like object;搜尋
b"ERROR",不是"ERROR"。 - 空檔案先檢查
st_size,不要直接建立長度 0 的 file mapping。 - 可寫 mapping 通常用
open("r+b")搭配ACCESS_WRITE。 - slice 原地替換必須維持相同長度。
- 部分 mapping 的 offset 必須向下對齊 allocation granularity。
- mapping 期間不要讓其他程序 truncate 檔案;必要時先建立快照。
- 用真實 access pattern 測量,不要拿一次性的 benchmark 代替設計判斷。
結語 #
mmap 最迷人的地方,不是把硬碟假裝成無限記憶體。
而是讓程式用熟悉的 bytes 操作,直接表達「我要看檔案的哪一段」。
今天的重點可以收成五句:
- 先從
ACCESS_READ與巢狀with開始。 - mapping 是 bytes-like object,搜尋與 regex 都要用 bytes。
- slice 仍可能複製資料,不要隨手
mm[:]。 - 原地修改適合固定長度欄位,長度改變就重寫新檔。
- 部分 mapping 的 offset 必須對齊
ALLOCATIONGRANULARITY。 下次遇到巨大 log、固定格式索引或已知 offset 的 binary file,不妨先問: 「我真的需要把整份檔案讀進 Python 嗎?」 有時候,讓作業系統幫忙翻頁,程式反而會更簡單。🔭