快轉到主要內容
  1. 教學文章/

Python mmap 實戰:記憶體映射、隨機存取與大型檔案搜尋

·7 分鐘· loading · loading · ·
Python Mmap Memory-Mapped-File Filesystem Performance Standard-Library
每日拍拍
作者
每日拍拍
科學家 X 科技宅宅
目錄
Python 學習 - 本文屬於一個選集。
§ 104: 本文

featured

一. 前言:檔案很大,不代表要整包讀進 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

關鍵是 PENDINGSUCCESS 都是 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=,通常比直接操作平台特定的 flagsprot 更容易跨平台。

七. 只映射一段: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 的隨機讀取 mmapseek()
跨 S3、HTTP、本機的統一介面 fsspec
長度會改變的內容重寫 新檔寫入後替換

mmap 適合多次查看大型檔案的不同位置、bytes 搜尋與固定布局資料;它不適合遠端 object storage、只讀一次的簡單 ETL,或長度會大幅變化的文字重寫。 處理本機與遠端儲存抽象時,請看 Python fsspec 實戰。 只想把路徑組合、遍歷與副檔名處理寫乾淨,則 Python pathlib 教學 更直接。

十. 常見陷阱清單
#

  1. mm[:]、大切片與 bytes(mm) 仍會建立副本,不是自動零複製。
  2. mapping 是 bytes-like object;搜尋 b"ERROR",不是 "ERROR"
  3. 空檔案先檢查 st_size,不要直接建立長度 0 的 file mapping。
  4. 可寫 mapping 通常用 open("r+b") 搭配 ACCESS_WRITE
  5. slice 原地替換必須維持相同長度。
  6. 部分 mapping 的 offset 必須向下對齊 allocation granularity。
  7. mapping 期間不要讓其他程序 truncate 檔案;必要時先建立快照。
  8. 用真實 access pattern 測量,不要拿一次性的 benchmark 代替設計判斷。

結語
#

mmap 最迷人的地方,不是把硬碟假裝成無限記憶體。 而是讓程式用熟悉的 bytes 操作,直接表達「我要看檔案的哪一段」。 今天的重點可以收成五句:

  1. 先從 ACCESS_READ 與巢狀 with 開始。
  2. mapping 是 bytes-like object,搜尋與 regex 都要用 bytes。
  3. slice 仍可能複製資料,不要隨手 mm[:]
  4. 原地修改適合固定長度欄位,長度改變就重寫新檔。
  5. 部分 mapping 的 offset 必須對齊 ALLOCATIONGRANULARITY。 下次遇到巨大 log、固定格式索引或已知 offset 的 binary file,不妨先問: 「我真的需要把整份檔案讀進 Python 嗎?」 有時候,讓作業系統幫忙翻頁,程式反而會更簡單。🔭

延伸閱讀
#

Python 學習 - 本文屬於一個選集。
§ 104: 本文

相關文章

Python shutil 實戰:檔案複製、搬移、壓縮與安全清理
·7 分鐘· loading · loading
Python Shutil Filesystem Automation Standard-Library Developer-Tools
Python tempfile 實戰:安全建立暫存檔案、目錄與測試資料
·9 分鐘· loading · loading
Python Tempfile Filesystem Testing Standard-Library Developer-Tools
Python unicodedata 實戰:文字正規化、搜尋與去重
·6 分鐘· loading · loading
Python Unicodedata Unicode Text-Normalization Search Standard-Library
Python shlex 實戰:安全拆解 Shell 參數、Quote 與迷你指令語法
·6 分鐘· loading · loading
Python Shlex Shell Cli Security Standard-Library
Python fsspec 實戰:統一讀寫本機、S3、HTTP 與資料管線路徑
·7 分鐘· loading · loading
Python Fsspec Filesystem S3 Data-Engineering ETL
Python socket 實戰:TCP client/server、timeout 與簡易通訊協定
·8 分鐘· loading · loading
Python Socket TCP Networking Standard-Library Developer-Tools