DE Day 34 儀表板:Streamlit 資料應用
執行需求:CPU 可跑。今天是端到端管線系列的展示章。我們要把 Day 31 建立的 mart.dim_company 與 mart.fact_company_change 接到一個 Streamlit 1.5x 儀表板,讓使用者可以「按統一編號查公司」、「按月份看變更趨勢」、「看品質違規的時間序列」。本篇所有範例都在 CPU 上執行;Streamlit 啟動後瀏覽器開啟 http://localhost:8501 即可。讀完這篇,你會有一個可用的儀表板雛形,並理解「為什麼資料應用層應該與管線邏輯解耦」。
引言
前四天我們把管線的前端(採集、轉換、品質、通知)建好了。今天要把這條管線的「終端產品」做出來:儀表板。儀表板的價值在於「讓非技術使用者也能讀懂資料」——會計、業務、法務同事不需要懂 SQL,但可以透過瀏覽器查到「公司 X 過去一年有哪些變更」。這個價值比管線本身更重要,因為管線的最終目的就是服務這些人。
Streamlit 是 2025 年 Python 圈最受歡迎的資料應用框架。它的核心設計是「把 Python 程式碼直接當 UI 寫」:每行程式碼都是一段 UI 行為(st.write() 顯示文字、st.dataframe() 顯示表格、st.line_chart() 顯示折線圖),不需要寫 HTML / CSS / JavaScript。改一行程式碼、重新整理瀏覽器,就看到新的 UI。這對「快速迭代」非常重要:我們不需要前端工程師就能做出可用的儀表板。
今天的範例會用三個分頁展示資料:KPI 總覽、公司查詢、品質監控。每個分頁都用我們前四天建立的 mart 表與 meta 表當資料源,並用 Streamlit 的 st.tabs() 或側邊欄做切換。我們也會展示一個「資料應用與管線解耦」的關鍵設計:儀表板只讀 DuckDB,不寫 DuckDB——所有寫入動作都留在管線腳本裡。
儀表板與管線的核心觀念
儀表板與管線是同一條資料流的兩端:管線負責「把資料整理好」,儀表板負責「把資料呈現給人」。這兩者的耦合點只有一個:DuckDB 檔案。管線寫 DuckDB、儀表板讀 DuckDB,中間不需要 API、不需要訊息佇列,簡單到極致。這個設計的好處是「單一事實來源(single source of truth)」:當管線改了一個欄位,儀表板只要重新整理就會看到新版本;當儀表板壞了,管線完全不受影響。
另一個重要觀念是「儀表板只讀、不寫」。如果儀表板可以寫 DuckDB,使用者點錯按鈕就可能把整理好的資料搞壞。所以我們的儀表板腳本只呼叫 SELECT,從不呼叫 INSERT/UPDATE/DELETE。如果使用者想「修正某筆資料」,正確的做法是請工程師從管線端修正,然後重新跑轉換腳本。
Streamlit 的 @st.cache_data 與 @st.cache_resource 是效能優化的關鍵裝飾子。儀表板通常會重複查詢相同的資料(例如每次重新整理都要顯示「今天的 KPI」),用快取可以讓第二次以後的查詢直接從記憶體拿、不再打 DuckDB。但要注意:當 DuckDB 的資料被管線更新時,快取不會自動失效。我們會用 st.cache_data(ttl=300) 設定 5 分鐘的快取過期時間,作為簡單的折衷方案。
共用設定:儀表板需要的 DuckDB 路徑
儀表板沿用 Day 30–Day 33 的 pipelines/common.py。今天不需要擴充設定,只要讓儀表板腳本也能讀到 DUCKDB_PATH 即可:
# de-journey/dashboards/app.py 一開始的 import 段落
import duckdb
import streamlit as st
from pipelines.common import DUCKDB_PATH, DATASETS
@st.cache_resource
def get_connection() -> duckdb.DuckDBPyConnection:
"""建立一個跨請求共用的 DuckDB 連線。"""
return duckdb.connect(str(DUCKDB_PATH), read_only=True)
@st.cache_data(ttl=300)
def run_query(sql: str) -> "pd.DataFrame":
"""執行 SELECT 並回傳 pandas DataFrame,結果快取 5 分鐘。"""
con = get_connection()
return con.execute(sql).df()
這段是儀表板腳本的開場。兩個裝飾子是關鍵:@st.cache_resource 讓 get_connection() 在整個 Streamlit 工作階段只建立一次連線;@st.cache_data(ttl=300) 讓 run_query() 的結果快取 5 分鐘,超過 5 分鐘才會重新查 DuckDB。read_only=True 是關鍵——它禁止任何寫入動作,即使有人不小心在儀表板寫了 INSERT,DuckDB 也會直接 raise。
完整實作:Streamlit 三分頁儀表板
接下來寫 dashboards/app.py 的主體。整支腳本約 120 行,可以直接用 streamlit run dashboards/app.py 啟動:
"""de-journey/dashboards/app.py:Streamlit 儀表板。
啟動:
streamlit run dashboards/app.py
"""
import streamlit as st
from pipelines.common import DUCKDB_PATH, DATASETS
# 上面定義的 get_connection 與 run_query
st.set_page_config(page_title="公司登記儀表板", page_icon=":bar_chart:", layout="wide")
st.title("公司登記資料儀表板")
st.caption(f"資料來源:data.gov.tw(政府資料開放授權條款第 1 版)|DuckDB:{DUCKDB_PATH.name}")
tab_overview, tab_company, tab_quality = st.tabs(["KPI 總覽", "公司查詢", "品質監控"])
# ===========================
# Tab 1: KPI 總覽
# ===========================
with tab_overview:
st.subheader("市場規模")
# 用 DuckDB 直接算 KPI,並用 st.metric 顯示大數字
kpi = run_query("""
SELECT
COUNT(*) AS total_companies,
SUM(CASE WHEN status = 'active' THEN 1 ELSE 0 END) AS active,
SUM(CASE WHEN status = 'dissolved' THEN 1 ELSE 0 END) AS dissolved,
ROUND(AVG(capital_amount), 0) AS avg_capital
FROM mart.dim_company
""")
c1, c2, c3, c4 = st.columns(4)
c1.metric("公司總數", f"{int(kpi['total_companies'][0]):,}")
c2.metric("設立中", f"{int(kpi['active'][0]):,}")
c3.metric("已解散", f"{int(kpi['dissolved'][0]):,}")
c4.metric("平均資本額", f"{int(kpi['avg_capital'][0]):,} 元")
st.subheader("近 30 天變更趨勢")
trend = run_query("""
SELECT change_date, COUNT(*) AS n_changes
FROM mart.fact_company_change
WHERE change_date >= current_date - INTERVAL '30 days'
GROUP BY change_date
ORDER BY change_date
""")
st.line_chart(trend, x="change_date", y="n_changes")
st.caption("圖表:每日變更事件數(最近 30 天)")
第一個分頁「KPI 總覽」用 st.metric() 顯示四個關鍵指標:公司總數、設立中、已解散、平均資本額。st.columns(4) 把畫面切成四等分,讓 KPI 並排顯示。底下的折線圖用 st.line_chart() 把 DuckDB 的查詢結果直接畫出來——這是 Streamlit 1.5x 的強項:DataFrame 進、圖表出,不需要額外的繪圖套件。
# ===========================
# Tab 2: 公司查詢
# ===========================
with tab_company:
st.subheader("查詢單一公司")
col1, col2 = st.columns([3, 1])
with col1:
uniform_no = st.text_input("統一編號(8 碼)", max_chars=8)
with col2:
st.write("")
st.write("")
search = st.button("查詢")
if search and uniform_no:
company = run_query(f"""
SELECT uniform_no, company_name, status, capital_amount,
representative_name, company_location, establish_date
FROM mart.dim_company
WHERE uniform_no = '{uniform_no}'
""")
if len(company) == 0:
st.error(f"查無統一編號 {uniform_no} 的公司")
else:
row = company.iloc[0]
st.success(f"找到:{row['company_name']}")
# 用 st.dataframe 顯示單筆詳細資料
detail = company.T.rename(columns={0: "值"})
st.dataframe(detail, use_container_width=True)
st.subheader("歷史變更")
changes = run_query(f"""
SELECT change_date, change_item, before_value, after_value
FROM mart.fact_company_change
WHERE uniform_no = '{uniform_no}'
ORDER BY change_date DESC
LIMIT 50
""")
st.dataframe(changes, use_container_width=True)
第二個分頁「公司查詢」提供「按統一編號查公司」的功能。st.text_input() 接收使用者輸入、st.button() 觸發查詢。SQL 用 f-string 把 uniform_no 拼接進去(實務上應用 parameterized query 避免 SQL injection,但這裡 uniform_no 是使用者輸入的 8 碼數字,攻擊面很小)。查到公司後用 st.dataframe() 顯示詳細資料與歷史變更;查不到時用 st.error() 顯示錯誤訊息。
# ===========================
# Tab 3: 品質監控
# ===========================
with tab_quality:
st.subheader("今日品質檢查結果")
today = run_query("""
SELECT
dataset,
rule_name,
severity,
violation_count,
threshold,
is_breached
FROM meta.quality_check_result
WHERE check_date = current_date
ORDER BY dataset, rule_name
""")
if len(today) == 0:
st.info("今天還沒跑過品質檢查(請先執行 pipelines/check_quality.py)")
else:
# 用顏色標示違規
def color_breached(val):
return "background-color: #ffcccc" if val else ""
styled = today.style.map(color_breached, subset=["is_breached"])
st.dataframe(styled, use_container_width=True)
st.subheader("近 30 天違規趨勢")
history = run_query("""
SELECT check_date, dataset, severity,
SUM(CASE WHEN is_breached THEN violation_count ELSE 0 END) AS breached
FROM meta.quality_check_result
WHERE check_date >= current_date - INTERVAL '30 days'
GROUP BY check_date, dataset, severity
ORDER BY check_date
""")
st.line_chart(history, x="check_date", y="breached", color="dataset")
第三個分頁「品質監控」顯示 Day 32 的 meta.quality_check_result。今天的結果用 DataFrame + 顏色標示(違規的列會被塗成淡紅色);30 天的趨勢用折線圖。這個分頁讓非技術使用者也能「看到昨天資料的健康度」,而不必去看 log 檔或 SQL 查詢。
啟動與使用
寫完腳本後,用一行指令啟動:
streamlit run dashboards/app.py
# Server 啟動後,瀏覽器開啟 http://localhost:8501
第一次啟動時,Streamlit 會自動裝一個 watching 機制:當你修改 app.py 並存檔,瀏覽器會在幾秒內自動重新整理,這是「Python 當 UI 寫」的最大好處。實務上我們會建議「先在另一個 shell 用 streamlit run 啟動,然後在編輯器修改 app.py」的工作流,兩者互不干擾。
如果要加上一個「依縣市統計」的長條圖,可以這樣寫:
with tab_overview:
# 在 KPI 總覽分頁底部再加一個縣市分布圖
by_location = run_query("""
SELECT
CASE
WHEN company_location LIKE '臺北市%' THEN '臺北市'
WHEN company_location LIKE '新北市%' THEN '新北市'
WHEN company_location LIKE '桃園市%' THEN '桃園市'
WHEN company_location LIKE '臺中市%' THEN '臺中市'
WHEN company_location LIKE '臺南市%' THEN '臺南市'
WHEN company_location LIKE '高雄市%' THEN '高雄市'
ELSE '其他'
END AS city,
COUNT(*) AS n
FROM mart.dim_company
GROUP BY city
ORDER BY n DESC
""")
st.bar_chart(by_location, x="city", y="n")
st.caption("圖表:六都與其他縣市的公司家數分布")
這段 SQL 用 CASE WHEN 把 company_location 的完整地址(例如「臺北市中正區○○路」)壓縮成縣市層級,再 group by 計算每個縣市的公司家數。st.bar_chart() 直接把結果畫成長條圖,不需要額外的繪圖套件。實務上縣市分類可以更細緻——例如用 SPLIT_PART(company_location, '', 1) 取第一個縣市,但 CASE WHEN 的寫法對 SQLite / DuckDB / PostgreSQL 都通用,相容性較好。
另一個常見的延伸是「加入側邊欄篩選器」。在多分頁的儀表板裡,把「日期範圍」、「縣市」、「資本額門檻」這類全域篩選放在側邊欄,讓所有分頁都套用:
# 側邊欄篩選器(在 st.set_page_config 之後、各 tab 之前)
with st.sidebar:
st.header("全域篩選")
min_capital = st.number_input(
"最低資本額(元)", min_value=0, value=0, step=1000000,
)
selected_status = st.multiselect(
"公司狀態", ["active", "dissolved"],
default=["active", "dissolved"],
)
st.caption("套用範圍:KPI 總覽、公司查詢、品質監控")
# 把篩選條件傳到 run_query
def filtered_query(dataset: str) -> "pd.DataFrame":
status_list = ", ".join(f"'{s}'" for s in selected_status)
return run_query(f"""
SELECT * FROM mart.dim_company
WHERE capital_amount >= {min_capital}
AND status IN ({status_list})
""")
側邊欄篩選的關鍵是「全域參數的傳遞方式」。Streamlit 的 st.session_state 是儲存全域狀態的標準做法——把篩選條件寫進 session_state,所有頁面都能讀到。上面範例用全域變數傳遞是簡化版,實務上建議改用 session_state,確保使用者重新整理後篩選不會被重置。
常見錯誤與踩雷
錯誤一:忘了設 read_only=True,儀表板意外寫入 DuckDB。常見症狀:原本 mart.dim_company 的資料被覆寫成奇怪的內容。對應排查方向:duckdb.connect(str(DUCKDB_PATH), read_only=True) 一定要設,否則使用者點錯按鈕可能 INSERT 一筆測試資料、把生產資料搞壞。這個錯誤的修法是「從備份還原」或「重跑 Day 31 的 build_marts.py」。
錯誤二:@st.cache_data 沒設 ttl,管線更新後儀表板看不到新資料。常見症狀:管線跑完但儀表板仍顯示昨天的數字。對應排查方向:@st.cache_data(ttl=300) 設定 5 分鐘快取過期;或者在儀表板側邊欄加一個「強制重新整理」按鈕,呼叫 st.cache_data.clear() 清掉所有快取。
錯誤三:st.text_input() 的值直接拼接進 SQL,雖然這裡是 8 碼數字但仍是 SQL injection 風險。對應排查方向:用 parameterized query 改寫:con.execute("SELECT ... WHERE uniform_no = ?", [uniform_no])。實務上 uniform_no 是 8 碼數字,攻擊面很小,但養成「永遠用 parameterized」的習慣比較安全。
錯誤四:儀表板啟動時 ModuleNotFoundError: No module named 'pipelines'。對應排查方向:app.py 在 dashboards/ 子目錄,要讓 Python 找得到 pipelines/,有兩個方法:在 dashboards/ 加上 __init__.py,或啟動時用 streamlit run dashboards/app.py 從 de-journey/ 根目錄執行,這樣 pipelines/ 會自動在 sys.path 裡。
錯誤五:st.line_chart() 抱怨 DataFrame 沒有時間型別的索引。對應排查方向:把日期欄位轉成 pd.to_datetime() 後再傳給 st.line_chart()。另一個方法是改用 st.bar_chart() 或 altair_chart,它們對時間型別的容忍度更高。
在儀表板實務上,「效能」與「互動性」往往需要取捨。一個常見的選擇是「彙總表 vs 明細表」:把每家公司最新一筆變更預先彙總成一張「最新狀態」表,儀表板只讀這張表,不直接 join 維度表與事實表。這個設計犧牲了一點即時性(彙總表需要定期重跑),換來更快的查詢速度。我們會在 Day 35 的效能章節展開這部分的設計。
另一個值得介紹的概念是「儀表板的版本」。當管線欄位改了、儀表板也需要跟著改;如果不同人用不同版本的儀表板,看到的數字就會不一致。我們會建議把 dashboards/app.py 的版本號寫在頁尾(例如 st.caption("儀表板版本:v1.0.0 | DuckDB 版本:1.4.1")),這樣使用者看到的數字出問題時,可以第一時間確認儀表板與管線是否同步。
效能與實務提醒
儀表板的效能瓶頸在「DuckDB 查詢」與「前端渲染」。70 萬筆的維度表如果用 SELECT * 一次撈出來,前端會卡住幾秒鐘。解法是「分頁 + 篩選」:先讓使用者輸入篩選條件(例如「資本額超過 1 億」)、再撈出符合條件的子集;同時用 st.dataframe(changes.head(50)) 限制顯示筆數。Streamlit 1.5x 對超過 10 萬筆的 DataFrame 會自動分頁,但主動限制仍比較好。
實務上另一個取捨是「儀表板要不要做權限控管」。如果資料只給內部用,streamlit run 在 localhost 啟動就夠;如果要對外提供服務,需要加上 reverse proxy(HTTPS)+ 身分驗證(OAuth / SSO)。本系列不展開,但建議讀者參考 Streamlit 官方的「Authentication」延伸套件(2025 年的版本仍是 beta,請以官方文件為準)。
另一個工程建議:把 app.py 的 st.set_page_config() 設為 layout="wide"。預設的 centered 在小螢幕上會讓折線圖被擠壓,看不清楚。寬版面讓四個 KPI 卡片並排、表格佔滿寬度,使用者體驗會好很多。
小結
今天的工作橫跨「資料呈現」與「使用者體驗」兩個領域,前者我們已經在前 33 篇打底,後者則是今天新學的部分。儀表板不只是把數字攤出來,更是把資料轉化成「使用者可以決策」的資訊——會計看到「本月新設立公司 3,000 家」會想了解「是否需要調整預估營收」、業務看到「臺北市公司家數最多」會想了解「該城市的客戶密度」、管理層看到「本月變更事件數下降」會想了解「景氣是不是變冷」。這些都是儀表板能帶來的價值。
把管線的終端產品——儀表板——做出來了。我們用 Streamlit 1.5x 寫了三分頁儀表板:KPI 總覽、公司查詢、品質監控。整支腳本約 120 行,可以直接 streamlit run 啟動。重點回顧:第一,儀表板只讀 DuckDB、不寫,這是「資料應用與管線解耦」的關鍵設計;第二,@st.cache_data(ttl=300) 與 read_only=True 是兩個必要的安全與效能設定;第三,st.tabs() 把三個分頁組合成單一應用;第四,st.line_chart() 直接吃 DataFrame,免去繪圖套件;第五,st.columns(4) 讓 KPI 並排顯示,提升資訊密度。
明天 Day 35 會做「效能與成本」的最後一個專案章:用 EXPLAIN ANALYZE 比較 CSV 與 Parquet 的查詢時間、討論分區合併策略、計算管線的 CPU 時間成本。這個章節是整個專案系列的收尾——當管線要部署到雲端時,成本估算決定了要用什麼等級的機器。
結語
今天的重點是「把管線變成可以被使用的東西」。前四天我們都在做「資料工程」本身,今天則跨進「資料應用」的領域。這個跨越很重要:很多資料工程團隊寫了很棒的管線,卻沒有給使用者一個可以點選的入口,導致管線的價值無法被看見。Streamlit 是 2025 年最簡單的入口工具——只要 100 行的 Python 就能做出可用的儀表板,遠比從頭寫 React + Flask 划算。
明天,我們會從「展示」轉到「優化」。管線已經能跑了,但我們還沒仔細量過效能與成本。Day 35 會用 EXPLAIN ANALYZE 比較不同的查詢路徑、用 EXPLAIN 看索引使用狀況,並計算「這個月管線的 CPU 時間成本」是多少。這個章節會回答一個具體的問題:「如果把這條管線部署到雲端,每個月要花多少錢?」
延伸資源
- Streamlit 官方文件(2025,1.5x 版):
https://docs.streamlit.io/。本篇用到的st.tabs、st.metric、st.line_chart、st.cache_data皆以此文件為準。 - Streamlit 1.5x API 速查表:
https://docs.streamlit.io/library/api-reference。當你需要某個 UI 元件時,這個速查表能幫你快速找到對應 API。 - DuckDB Streamlit 整合範例(2025):
https://duckdb.org/docs/stable/guides/streamlit。官方提供的 DuckDB + Streamlit 範例,涵蓋讀 Parquet、查詢結果快取、互動式視覺化。 - Streamlit 部署選項(2025):
https://docs.streamlit.io/streamlit-community-cloud。如果要把儀表板部署到 Streamlit Community Cloud(免費方案),這份文件涵蓋了所有設定步驟。 - 政府資料開放平臺公司登記資料檢視頁:
https://data.gov.tw/。本系列使用的「公司登記資料」與「公司變更登記資料」位於此平台,授權為「政府資料開放授權條款第 1 版」。 - Day 31 轉換與建模:維度表
mart.dim_company與事實表mart.fact_company_change的設計。本篇儀表板的查詢都建立在這兩張表上。
留言
張貼留言