一. 前言：資料查詢不一定要先架資料庫 #
很多團隊都有一種很微妙的資料需求： 資料不是大到需要一整套 data warehouse， 也不是小到用 Excel 手拉一拉就結束。 它可能是一包每天更新的 CSV， 可能是一批 Parquet 檔， 也可能是某個排程丟在 shared folder 的報表。 這種時候常見做法有三種：
- 用 pandas 寫 notebook。
- 把資料匯進 PostgreSQL。
- 做一個臨時 Streamlit Dashboard。 三種都可以。 但如果你只是想讓同事在瀏覽器裡選日期、選類別、看結果、下載 CSV， 完整資料庫有時候太重，notebook 又不太像給別人操作的工具。 今天拍拍君要介紹一個務實組合： Streamlit 負責介面，DuckDB 負責查詢。 Streamlit 讓你用 Python 寫出互動式 Web App。 DuckDB 讓你直接用 SQL 查 CSV、Parquet 和 DataFrame。 兩個接起來，就能做一個本地資料查詢 Dashboard：
- 使用者選擇資料檔。
- UI 控制日期、地區、類別。
- DuckDB 用參數化 SQL 查詢。
- Streamlit 顯示表格、指標、圖表和下載按鈕。 如果你還沒看過前面的文章，可以先補： Streamlit 入門篇、 Streamlit 進階篇、 Python DuckDB 實戰。 這篇不重講 Streamlit 基礎元件，也不重講 DuckDB 怎麼查第一個 CSV。 我們直接做一個接近實務的小工具。
二. 什麼情境適合這個組合？ #
Streamlit + DuckDB 特別適合「本地分析資料 App」。 例如：
- 每天產生一批銷售 CSV，需要快速查詢。
- 有很多 Parquet 檔，想給非工程同事篩選。
- 內部工具要查 log summary，但還不值得建資料庫。
- 資料科學家想把分析結果包成小 App。
- 你想先驗證需求，再決定要不要升級成正式系統。 DuckDB 的好處是它不需要 server。 你不用開 PostgreSQL，不用設帳號，不用處理 migration。 資料可以留在檔案裡，查詢時才讀。 Streamlit 的好處是 UI 成本很低。 你不用寫前端框架，也不用先設計 API。 一支 Python script 就能開始。 但它也不是萬能。 不適合的情境：
- 需要多人同時大量寫入資料。
- 需要細緻權限管理與審計紀錄。
- 資料必須長期穩定交易一致性。
- Dashboard 需要很複雜的前端互動。
- 查詢結果要支援很高併發。 簡單說： 這是內部工具、資料探索、輕量查詢介面的甜蜜點。 等需求真的長大，再把查詢層搬到正式資料庫也不遲。
三. 建立專案 #
先建立一個小專案。
uv init streamlit-duckdb-dashboard
cd streamlit-duckdb-dashboard
uv add streamlit duckdb pandas pyarrow
不用 uv 的話：
python -m venv .venv
source .venv/bin/activate
pip install streamlit duckdb pandas pyarrow
今天的專案結構長這樣：
streamlit-duckdb-dashboard/
├── app.py
└── data/
└── orders.csv
先準備一份範例資料，存成 data/orders.csv：
order_id,order_date,customer,region,product,category,quantity,unit_price
1001,2026-05-01,拍拍醬,North,Notebook,Stationery,3,6.5
1002,2026-05-01,chatPTT,South,Keyboard,Hardware,1,79.0
1003,2026-05-02,拍拍君,East,Mouse,Hardware,2,25.0
1004,2026-05-03,小拍,North,Coffee,Food,5,4.2
1005,2026-05-04,拍拍醬,West,Monitor,Hardware,1,199.0
資料不大沒關係。 重點是先把 Dashboard 的查詢模式做對。 之後換成幾十萬列 CSV 或 Parquet，架構也差不多。
四. 最小 Streamlit App #
先做一個能跑起來的版本。
from pathlib import Path
import duckdb
import streamlit as st
DATA_DIR = Path("data")
DEFAULT_FILE = DATA_DIR / "orders.csv"
st.set_page_config(
page_title="拍拍君資料查詢 Dashboard",
page_icon="📊",
layout="wide",
)
st.title("拍拍君資料查詢 Dashboard")
st.caption("Streamlit + DuckDB：用 SQL 查本地 CSV / Parquet")
df = duckdb.sql(f"SELECT * FROM read_csv_auto('{DEFAULT_FILE}')").df()
st.dataframe(df, use_container_width=True)
啟動：
uv run streamlit run app.py
這個版本已經證明 Streamlit 可以啟動，DuckDB 也能讀到 CSV。 但它還有幾個問題：
- 檔案路徑直接塞進 SQL 字串。
- 沒有快取，每次互動都可能重新讀檔。
- 沒有篩選條件。
- 使用者還不能下載查詢結果。 接下來我們把它修成比較可靠的版本。
五. 把 DuckDB 查詢包起來 #
先建立一個小函式，把檔案格式和查詢集中管理。
from pathlib import Path
import duckdb
import pandas as pd
def table_expr_for(path: Path) -> str:
suffix = path.suffix.lower()
if suffix == ".csv":
return "read_csv_auto(?)"
if suffix == ".parquet":
return "read_parquet(?)"
raise ValueError("只支援 CSV 與 Parquet")
def read_preview(path: Path) -> pd.DataFrame:
table_expr = table_expr_for(path)
sql = f"SELECT * FROM {table_expr} LIMIT 1000"
with duckdb.connect() as con:
return con.execute(sql, [str(path)]).df()
這裡有兩個重點。
第一，檔案格式走白名單。
table_expr_for() 只會回傳 read_csv_auto(?) 或 read_parquet(?)。
第二，檔案路徑走參數。
我們用 ?，而不是把路徑直接拼進 SQL。
DuckDB 的 Python API 支援：
con.execute("SELECT * FROM read_csv_auto(?)", [path])
這比自己處理引號安全，也比較不容易被特殊字元弄壞。
六. 加上 Streamlit 快取 #
Streamlit 的執行模型是：
使用者每次互動，整支 script 重新跑一次。
這很直覺，但資料讀取要小心。
如果每次滑動篩選器都重新掃一次大檔案，Dashboard 很快就會變慢。
讀檔結果適合用 st.cache_data。
@st.cache_data(show_spinner="讀取資料中...")
def load_preview(path: str) -> pd.DataFrame:
df = read_preview(Path(path))
df["order_date"] = pd.to_datetime(df["order_date"])
return df
注意參數用 str。
Streamlit cache 需要把參數雜湊起來，簡單型別最穩。
如果檔案內容會變，但路徑不變，可以加上 TTL：
@st.cache_data(ttl=300)
def load_preview(path: str) -> pd.DataFrame:
...
或提供一個重新讀取按鈕：
if st.sidebar.button("重新讀取資料"):
st.cache_data.clear()
這比叫使用者重開 App 友善多了。
七. 用 UI 條件組安全查詢 #
先從 preview 裡取得可用條件。
preview = load_preview(str(DEFAULT_FILE))
min_date = preview["order_date"].min().date()
max_date = preview["order_date"].max().date()
date_range = st.sidebar.date_input(
"訂單日期",
value=(min_date, max_date),
min_value=min_date,
max_value=max_date,
)
regions = tuple(sorted(preview["region"].dropna().unique()))
categories = tuple(sorted(preview["category"].dropna().unique()))
selected_regions = tuple(
st.sidebar.multiselect("地區", regions, default=regions)
)
selected_categories = tuple(
st.sidebar.multiselect("類別", categories, default=categories)
)
很多範例會直接用 pandas filter。 這當然可以。 但今天我們希望 DuckDB 負責查詢，因為：
- SQL 條件更接近未來資料庫查詢。
- DuckDB 可以直接查 Parquet，不一定要先整份讀成 DataFrame。
- 複雜聚合、排序、分組用 SQL 比較清楚。 接著寫查詢函式。
@st.cache_data(show_spinner="查詢資料中...")
def run_query(
path: str,
start_date,
end_date,
regions: tuple[str, ...],
categories: tuple[str, ...],
) -> pd.DataFrame:
source = Path(path)
table_expr = table_expr_for(source)
where = ["CAST(order_date AS DATE) BETWEEN ? AND ?"]
params: list[object] = [str(source), start_date, end_date]
if regions:
where.append(f"region IN ({', '.join(['?'] * len(regions))})")
params.extend(regions)
if categories:
where.append(f"category IN ({', '.join(['?'] * len(categories))})")
params.extend(categories)
sql = f"""
SELECT
order_id,
CAST(order_date AS DATE) AS order_date,
customer,
region,
product,
category,
quantity,
unit_price,
quantity * unit_price AS revenue
FROM {table_expr}
WHERE {" AND ".join(where)}
ORDER BY order_date, order_id
"""
with duckdb.connect() as con:
return con.execute(sql, params).df()
這段有一個務實規則：
SQL 結構可以由程式白名單組合；使用者資料要走參數。
table_expr 來自我們自己的白名單，所以可以放進 f-string。
日期、地區、類別來自 UI，所以全部放進 params。
如果 date_input 回傳範圍，就這樣處理：
if len(date_range) == 2:
start_date, end_date = date_range
else:
start_date = end_date = date_range[0]
然後執行：
result = run_query(
str(DEFAULT_FILE),
start_date=start_date,
end_date=end_date,
regions=selected_regions,
categories=selected_categories,
)
八. 顯示指標、圖表與表格 #
空結果要先處理，不然後面的圖表可能會出錯。
if result.empty:
st.warning("目前篩選條件沒有資料。")
st.stop()
接著顯示三個指標。
total_revenue = result["revenue"].sum()
order_count = len(result)
avg_revenue = total_revenue / order_count
col1, col2, col3 = st.columns(3)
col1.metric("訂單數", f"{order_count:,}")
col2.metric("總營收", f"${total_revenue:,.2f}")
col3.metric("平均訂單金額", f"${avg_revenue:,.2f}")
再加一個每日營收圖。
daily = (
result.groupby("order_date", as_index=False)["revenue"]
.sum()
.sort_values("order_date")
)
st.line_chart(daily, x="order_date", y="revenue")
st.dataframe(result, use_container_width=True, hide_index=True)
這裡用 pandas groupby 是可以接受的。 資料已經被 DuckDB 篩小了。 拍拍君通常會這樣分工：
- 大量資料篩選、join、group by：DuckDB。
- 查詢後的小型結果整理：pandas。
- 顯示、互動、下載：Streamlit。 這樣程式會比較清楚。
九. 下載、限制與安全邊界 #
Dashboard 很常被要求「可以下載目前結果嗎？」
這可以直接用 st.download_button()：
st.download_button(
"下載查詢結果 CSV",
data=result.to_csv(index=False).encode("utf-8-sig"),
file_name="orders-filtered.csv",
mime="text/csv",
)
utf-8-sig 是為了讓 Excel 比較不容易亂碼。
如果使用者是 Python / R / Spark，可以再加一個 Parquet 下載。
但下載之前最好做限制：
if len(result) > 100_000:
st.warning("結果超過 100,000 列，請縮小篩選範圍再下載。")
st.stop()
另一個常見誘惑，是加一個文字框讓使用者任意輸入 SQL。 自己本機用可以；開給同事或內部團隊用，拍拍君會先踩煞車。 任意 SQL 代表使用者可能查到你沒打算暴露的欄位、做超重查詢拖慢 App，甚至讀取本機上其他可見路徑。 比較安全的做法是提供固定篩選元件、有限排序選項、白名單欄位選擇，把 SQL 結構控制在程式裡。 例如排序欄位可以這樣白名單化：
sort_options = {"日期": "order_date", "營收": "revenue", "數量": "quantity"}
sort_column = sort_options[st.sidebar.selectbox("排序欄位", list(sort_options))]
sql = f"SELECT * FROM filtered_orders ORDER BY {sort_column} DESC"
這裡的 f-string 可以接受，因為 sort_column 來自白名單 dict，不是使用者任意字串。
最後記住幾個坑：
- 不要每次互動都重新掃大檔案；用
st.cache_data。 - 優先讓 DuckDB 篩選，再把結果轉成 DataFrame。
- 使用者輸入不要直接拼進 SQL。
- 下載結果要限制大小。
- Streamlit + DuckDB 是輕量查詢工具，不是完整資料平台。 當使用者開始要求登入、角色權限、審計紀錄、多服務資料來源，或查詢結果需要被其他系統呼叫，就該考慮 PostgreSQL / ClickHouse + FastAPI 這類正式架構。
結語：先把資料變成可以操作的工具 #
很多資料工作卡住，不是因為模型不夠強，也不是因為資料庫不夠大。 而是資料還停在「只有寫程式的人能看懂」的狀態。 Streamlit + DuckDB 的組合很適合打破這個卡點。 DuckDB 讓你用 SQL 快速查本地檔案。 Streamlit 讓你把查詢變成普通人可以操作的介面。 兩個接起來，你就能用很低的成本做出實用的資料查詢 Dashboard。 今天這篇的核心不是多炫的圖表。 而是幾個務實原則：
- 查詢條件要參數化。
- 檔案格式與動態 SQL 結構要白名單。
- 大資料讀取要 cache。
- 先篩選，再顯示，再下載。
- 小工具先解決真問題，長大後再升級架構。 下次你看到一包 CSV 或 Parquet，不一定要先開 notebook，也不一定要先架資料庫。 可以先問問看： 「這是不是一個 Streamlit + DuckDB 就能解決的查詢 App？」 很多時候，答案會是 yes。 而且會比你想像中快。拍拍。