Streamlit 效能診斷：Rerun、Cache Hit、記憶體與 Latency Profiling

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

§ 129: 本文

Streamlit App 變慢時，最危險的一句話是：「應該是圖表太多吧。」

也可能是每次 widget 互動都重讀檔案、cache key 不斷改變、某個 API 偶爾卡住，或同一份 DataFrame 被複製到記憶體好幾次。

如果沒有量測，優化就只是在猜。

這篇拍拍君不再重教 st.cache_data 怎麼寫，也不把一般 Python profiler 全搬過來。我們要建立一套 Streamlit 專用的診斷流程：

先定義冷啟動與暖 rerun 的 latency budget；
量測整輪 rerun，以及 load、transform、render 各階段；
用受控實驗分辨 cache hit、miss 與 key 爆炸；
同時觀察 RSS 和 Python heap，不被單一數字騙走；
用 AppTest 留下可重複比較的 server-side benchmark。

如果你想先補 Streamlit 的 rerun 與 cache 基礎，請看 Streamlit 進階篇；想縮小 rerun 範圍，則接著看 Fragments + Dialogs。

一. 先定義「慢」：不要只說感覺
#

效能問題至少要分成四條路徑：

路徑	使用者動作	主要風險	範例 budget
Cold start	第一次開頁	import、連線、讀檔、cache miss	p95 ≤ 2.5 s
Warm rerun	改一個篩選	重算、重畫、無效 I/O	p95 ≤ 350 ms
Submit	送出表單	查詢、模型、外部 API	p95 ≤ 1.2 s
Long session	連續操作 30 次	cache 膨脹、物件滯留	RSS 增幅 ≤ 150 MB
這些不是宇宙真理，只是範例。

真正重要的是把路徑、百分位數、測試資料量與環境寫清楚。平均 200 ms 可能藏著每十次就有一次 3 秒的尖峰；p95 或 p99 才比較接近使用者抱怨的那一刻。

也要分清三種時間：

Python 執行時間：server 跑 script 與函式的時間；
傳輸時間：server 把資料與 UI delta 送到瀏覽器的時間；
瀏覽器呈現時間：前端真正畫完表格或圖表的時間。

本文主要量第一種。AppTest 也不會啟動真瀏覽器，所以不能假裝它量到了完整 click-to-paint latency。

二. 建立可重現的診斷專案
#

先做一個小型銷售 Dashboard：

mkdir pypy-streamlit-perf
cd pypy-streamlit-perf
uv init
uv add streamlit pandas numpy psutil
uv add --dev pytest
mkdir -p data tests

固定環境資訊：

uv run python --version
uv run streamlit version
uv pip freeze > environment.txt

不要拿開發機的 1,000 列資料和正式環境的 300 萬列資料比較。

範例用固定 seed 產生資料：

# make_data.py
from pathlib import Path

import numpy as np
import pandas as pd

rng = np.random.default_rng(42)
rows = 300_000

df = pd.DataFrame(
    {
        "region": rng.choice(["北", "中", "南", "東"], rows),
        "product": rng.choice(["拍拍咖啡", "拍拍餅乾", "拍拍筆記"], rows),
        "revenue": rng.integers(100, 10_000, rows),
        "units": rng.integers(1, 20, rows),
    }
)
Path("data").mkdir(exist_ok=True)
df.to_parquet("data/sales.parquet", index=False)

uv run python make_data.py

每次 benchmark 都用同一份資料、同一版依賴與相近的機器負載。否則差異可能來自資料或環境，不是你的改動。

三. 先畫出 Rerun 路徑
#

Streamlit 預設在使用者互動後重新執行 script。診斷前先列出每輪會經過的工作：

widget change
  ├─ import / page setup
  ├─ load_sales()
  ├─ filter_sales()
  ├─ aggregate_sales()
  ├─ build_chart_data()
  └─ render metrics / chart / dataframe

接著逐項問：

這一步每次都需要嗎？
輸入真的變了嗎？
結果可以快取嗎？
結果有多大？
時間花在 CPU、I/O，還是序列化？

先別急著加 cache。錯誤的 cache 只會把問題從 latency 變成記憶體，還可能因 key 太多而越跑越胖。

四. 用 `perf_counter()` 量每個階段
#

time.perf_counter() 適合量 elapsed time，因為它是單調遞增的高解析度時鐘。

建立一個簡單 probe：

# perf_probe.py
from __future__ import annotations

from contextlib import contextmanager
from time import perf_counter
from typing import Iterator


@contextmanager
def measure(name: str, sink: list[dict[str, float | str]]) -> Iterator[None]:
    started = perf_counter()
    try:
        yield
    finally:
        sink.append(
            {
                "stage": name,
                "elapsed_ms": (perf_counter() - started) * 1_000,
            }
        )

在 App 裡把每輪資料放進新的 list：

# app.py
from time import perf_counter

import pandas as pd
import streamlit as st

from perf_probe import measure

run_started = perf_counter()
timings: list[dict[str, float | str]] = []


@st.cache_data(show_spinner=False)
def load_sales(path: str) -> pd.DataFrame:
    return pd.read_parquet(path)


with measure("load", timings):
    sales = load_sales("data/sales.parquet")

region = st.selectbox("地區", ["全部", "北", "中", "南", "東"])

with measure("filter", timings):
    filtered = sales if region == "全部" else sales[sales["region"] == region]

with measure("aggregate", timings):
    summary = filtered.groupby("product", as_index=False)["revenue"].sum()

with measure("render", timings):
    st.bar_chart(summary, x="product", y="revenue")
    st.dataframe(filtered.head(500), use_container_width=True)

total_ms = (perf_counter() - run_started) * 1_000
st.caption(f"server-side rerun: {total_ms:.1f} ms")

with st.expander("本輪診斷"):
    st.dataframe(timings, hide_index=True)

這份數字回答的是「Python 呼叫 st.dataframe() 花多久」，不是瀏覽器畫完 500 列花多久。

不過它已經能抓出很多問題，例如 load 每次都 800 ms，或 aggregate 隨篩選條件暴增。

不要讓量測本身變成噪音
#

正式 benchmark 建議：

關掉 debug log 與不必要的 st.write()；
至少跑 10～30 次，不要只看一次；
分開記錄 cold 與 warm；
用 median、p95，而不是挑最好看的數字；
診斷面板不要塞完整 DataFrame 或大量物件。

五. Cache Hit / Miss：做受控實驗
#

Streamlit 的 st.cache_data 在輸入與函式程式碼未變時重用結果。show_time=True 可以在 cache miss 計算期間顯示耗時，但它不是完整監控系統。

@st.cache_data(
    ttl="15m",
    max_entries=8,
    show_spinner="讀取資料中…",
    show_time=True,
)
def load_sales(path: str, mtime_ns: int) -> pd.DataFrame:
    return pd.read_parquet(path)

把檔案修改時間放進 key，資料更新時就會 miss：

from pathlib import Path

path = Path("data/sales.parquet")
sales = load_sales(str(path), path.stat().st_mtime_ns)

要驗證 hit / miss，不要靠眼睛猜。設計三輪：

load_sales.clear() 後以固定參數呼叫，這是受控 cold miss；
相同參數再呼叫，這是 warm hit；
改 mtime_ns 或查詢參數，確認預期 miss。

from time import perf_counter


def timed_call(label: str, func, *args) -> dict[str, float | str]:
    started = perf_counter()
    func(*args)
    return {
        "case": label,
        "elapsed_ms": (perf_counter() - started) * 1_000,
    }


load_sales.clear()
results = [
    timed_call("cold", load_sales, str(path), path.stat().st_mtime_ns),
    timed_call("warm", load_sales, str(path), path.stat().st_mtime_ns),
]
st.dataframe(results, hide_index=True)

這段只適合本機診斷頁，不要把「清掉所有人的全域 cache」按鈕隨便放進 production。

Cache key 爆炸比 miss 更麻煩
#

下面的函式可能為每個自由輸入字串建立一份結果：

@st.cache_data
def search_sales(query: str) -> pd.DataFrame:
    ...

使用者輸入 p、py、pyp、pypy，就可能形成四個 key。修法不一定是「多加 cache」，而可能是：

用 st.form 等使用者 submit；
正規化大小寫與空白；
限制 max_entries；
只 cache 穩定的 load / aggregate 邊界；
不 cache 很小、很快、key 又很多的 filter。

st.cache_data 回傳資料副本；大型 DataFrame 的反序列化與複製仍然有成本。Hit 不代表免費。

六. 記憶體：RSS 與 Python Heap 要一起看
#

RSS 是作業系統看到的 process resident memory，會包含 Python heap、NumPy / Arrow 原生配置、載入的 library 與其他 session。

# memory_probe.py
import os

import psutil


def rss_mb() -> float:
    process = psutil.Process(os.getpid())
    return process.memory_info().rss / 1024 / 1024

在重要階段前後取樣：

before = rss_mb()
sales = load_sales(str(path), path.stat().st_mtime_ns)
after_load = rss_mb()
filtered = sales[sales["region"] == region]
after_filter = rss_mb()

st.json(
    {
        "rss_before_mb": round(before, 1),
        "rss_after_load_mb": round(after_load, 1),
        "rss_after_filter_mb": round(after_filter, 1),
    }
)

單次差值可能受 allocator、垃圾回收與其他 session 干擾。真正有用的是固定流程的趨勢：

啟動後記 baseline；
完成一次 cold load；
用 20 組不同篩選條件操作；
回到原條件並等待；
比較 RSS 是否持續單調上升。

若只想看 Python 配置，可以用 tracemalloc：

import tracemalloc

tracemalloc.start()
snapshot_before = tracemalloc.take_snapshot()

filtered = sales[sales["region"] == region].copy()

snapshot_after = tracemalloc.take_snapshot()
for stat in snapshot_after.compare_to(snapshot_before, "lineno")[:5]:
    print(stat)

但 tracemalloc 看不到所有 NumPy、Pandas、Arrow 或 C extension 的原生記憶體，所以它和 RSS 是互補，不是二選一。

`cache_resource` 要特別小心
#

模型、資料庫連線與 client 適合 st.cache_resource。全域 resource 會跨 session 共用，而且必須 thread-safe；它不是「比較省記憶體的 cache_data」。

診斷時要記錄：

建立 resource 前後的 RSS；
resource 是否真的只建立一次；
輸入參數是否意外產生多份 resource；
TTL 或淘汰後是否需要清理外部資源；
多使用者同時操作時是否安全。

七. 用 AppTest 建立 Rerun 基準
#

Streamlit AppTest 可以模擬 app、修改 widget，再明確呼叫 .run()。我們可以在 pytest 外層量每次 server-side run：

# tests/test_performance.py
from statistics import median
from time import perf_counter

from streamlit.testing.v1 import AppTest


def run_ms(app: AppTest) -> float:
    started = perf_counter()
    app.run(timeout=10)
    return (perf_counter() - started) * 1_000


def percentile(values: list[float], fraction: float) -> float:
    ordered = sorted(values)
    index = round((len(ordered) - 1) * fraction)
    return ordered[index]


def test_warm_region_reruns_stay_inside_budget():
    app = AppTest.from_file("app.py", default_timeout=10)
    cold_ms = run_ms(app)

    samples = []
    for region in ["北", "中", "南", "東"] * 5:
        app.selectbox("地區").set_value(region)
        samples.append(run_ms(app))

    assert not app.exception
    assert median(samples) < 300
    assert percentile(samples, 0.95) < 500
    print({"cold_ms": cold_ms, "warm_ms": samples})

執行：

uv run pytest -q -s tests/test_performance.py

AppTest 的好處是可重複、容易進 CI；限制是沒有真正瀏覽器、網路往返與前端繪圖成本。

因此它適合抓「某次重構讓 server rerun 從 180 ms 變成 900 ms」，不適合宣稱「使用者一定在 500 ms 內看到畫面」。

共享 CI runner 的負載也會抖。與其把 threshold 設得只比目前數字多 1%，不如：

固定資料與依賴；
留合理 headroom；
連續多次超標才阻擋；
保存 raw samples，讓人能判斷是回歸還是 runner 抖動。

功能測試的完整寫法可接著看 Streamlit AppTest + pytest。

八. 一份實用診斷順序
#

遇到慢 App 時，拍拍君會照這個順序：

重現：固定資料、操作路徑、版本與機器。
分路徑：cold、warm、submit、long session 分開量。
量總時間：先確認問題真的存在，而且能重複。
拆階段：load、transform、query、render 分別計時。
做 cache 實驗：clear → cold → warm → changed key。
看記憶體趨勢：RSS 搭配 tracemalloc，不迷信單點。
一次改一件事：否則不知道哪個修改有效。
重新量測：保存 before / after raw samples。
加回歸基準：用 AppTest 或更完整的瀏覽器測試守住成果。

常見判讀方式：

現象	優先檢查
Cold 慢、warm 快	import、resource 初始化、預熱策略
Cold 與 warm 都慢	cache key、未快取 I/O、每輪重算
Server 快、使用者仍覺得慢	傳輸量、表格列數、瀏覽器 render
RSS 隨不同輸入上升	key cardinality、大型 cache、物件滯留
p50 正常、p95 很差	外部 API、鎖競爭、偶發 cache miss
每次輸入一字就變慢	widget rerun、自由文字 key、缺少 form
如果最後定位到一般 Python 函式內部，再用 cProfile + line_profiler 深挖；順序不要倒過來。