第 6 章：讓重跑不再產生重複 — 去重與冪等（upsert）

上一章你發現：重跑 producer，資料就翻倍。這一章用一個資料工程的核心觀念「冪等」把它解決掉。

---

做完這一章，你會做到

1. 說得出「冪等（idempotent）」是什麼，並知道為什麼定期抓取一定要冪等。
2. 看懂怎麼用「複合主鍵 + on_duplicate_key_update（upsert）」做到去重。
3. 驗證：這一版無論重跑幾次，資料庫筆數都不變。

---

先搞懂：什麼是「冪等」

打個比方：電梯的「5 樓」按鈕，你按一次是去 5 樓，連按五次還是去 5 樓——多按沒有副作用。這就是冪等：同一個操作做幾次，結果都一樣。

上一章的寫法不冪等：每跑一次就 append 一批，跑三次就三份重複。但真實情況每天都會重抓、失敗會重試、補資料會重跑，不冪等的 pipeline 會越跑越髒。

解法：讓「同一筆資料」有一個唯一身分（主鍵），寫入時如果這筆已經存在就更新、不存在才新增。這個「有就更新、沒有就新增」的動作叫 upsert（update + insert）。

---

這一章會用到的檔案

檔案	角色	說明
crawler/tasks_crawler_finmind_duplicate.py	任務定義	crawler_finmind_duplicate + upload_data_to_mysql_duplicate
crawler/producer_crawler_finmind_duplicate_single.py	生產者（本章示範用）	用 .delay() 派送、不指定佇列——本章只看去重效果，一次只看一件事
crawler/producer_crawler_finmind_duplicate.py	生產者（分流版）	同樣的任務走 twse/tpex 多佇列分流（第 3 章的路由）

---

DataFrame 與資料清理

等下的 upsert 程式碼會逐列跑過一個 DataFrame，所以先認識它。

什麼是 DataFrame：

• DataFrame 是 Python 套件 pandas 的核心資料結構，可以想像成「強化版的 Excel 表格」
• 由**列（rows）和欄（columns）**組成，支援不同型態的資料（數字、文字、時間…）
• 專為資料處理設計，能幫你：讀取各種資料來源 (read_sql、read_csv)、檢查與清理資料品質、篩選/轉換/聚合、輸出結果（to_sql、to_csv）

空值處理的原則：

• 爬蟲時就做好防呆、預設值處理——API 回來的資料可能缺欄位、缺值，越早處理越省事
• 資料還是 list of dict 時要逐筆處理，比較沒效率
• 轉成 DataFrame 就能統一處理：查看/處理單一欄位、新增/刪除欄位、篩選查詢

資料怎麼進 DataFrame：三個入口

入口	語法	本課哪裡用
API 回傳（list of dict）	pd.DataFrame(data)	爬蟲任務（下面「線上」三步）
CSV 檔	pd.read_csv(...)	第 2 章存下的 output/TaiwanStockPrice_2330.csv
資料庫	pd.read_sql(SQL, engine)	把第 5 章寫進去的 TaiwanStockPrice 讀回來

from sqlalchemy import create_engine
engine = create_engine("mysql+pymysql://root:1234@127.0.0.1:3306/mydb")

df_csv = pd.read_csv("output/TaiwanStockPrice_2330.csv")        # CSV 檔 → df
df_db  = pd.read_sql("SELECT * FROM TaiwanStockPrice", engine)  # 資料庫的表 → df

to_sql 和 read_sql 是一對：第 5 章用 to_sql 把 df 寫進 MySQL，這裡用 read_sql 把表讀回成 df——資料在「DataFrame ↔ 資料庫」之間雙向流動。

型別注意：read_sql 讀回來的欄位型別由「資料庫欄位的型別」決定。第 5 章 to_sql 自動建表時把 stock_id 猜成了整數（BIGINT），所以讀回來是 int64、0050 變成 50——df_db[df_db["stock_id"] == "2330"] 用字串比對會得到 0 筆。要嘛比對前先 astype(str)，要嘛一開始就自己定義表結構（本章後半做的事）。

線下：先用 Jupyter 熟悉 DataFrame

repo 練習腳本： example/pandas.ipynb

Notebook 的內容地圖（用「故意留了空值」的學生資料，每個方法都看得到效果）：

段落	練什麼
基本資訊	head() / shape / dtypes / info() / describe() / isna().sum()
篩選資料	單欄、多欄、條件篩選 df[df["score"] > 80]、iloc
資料轉換	新增欄位、astype 改型別（含允許缺值的 Int64）
空值處理	找缺值的列、fillna(0) / 填平均值 / 填固定日期、dropna()
檔案輸出	to_csv / read_csv
練習	訂單資料 4 題（檢查 → 補值 → 篩選排序 → 新增欄位運算）
真實股價段	下面「線上」的同一套流程，在 notebook 裡自己跑一次

notebook 股價段還包含「加工」練習——新增衍生欄位與篩選：

df["spread_pct"] = (df["spread"] / df["open"] * 100).round(2)  # 新增欄位：漲跌幅 %
df = df.drop(columns=["Trading_turnover"])                     # 刪掉用不到的欄
hot = df[df["Trading_Volume"] > 30_000_000]                    # 篩選：成交量 > 3 千萬股
# 輸出：186 / 271 天；例如 2025-06-04 close=990.0、spread_pct=+4.11

注意：這類衍生欄位（例如漲跌幅）只在分析時計算，不寫進資料庫——資料庫存 API 的原始值就好。第 8 章會在 Metabase 用自訂欄位算，第 15 章會在 BigQuery 用 View 算均線。

線上：跟任務程式相同的三步

import requests, pandas as pd

# ① 跟爬蟲任務同一套：打 FinMind API，拿回 list of dict
r = requests.get("https://api.finmindtrade.com/api/v4/data", params={
    "dataset": "TaiwanStockPrice", "data_id": "2330",
    "start_date": "2025-06-01", "token": ""})
data = r.json()["data"]
# 拿到 271 筆，第一筆長這樣：
# {'date': '2025-06-02', 'stock_id': '2330', 'Trading_Volume': 40608468, ..., 'close': 946.0, 'spread': -21.0}

df = pd.DataFrame(data)      # ② 一行：list of dict → 表格
df.head(3)
#          date stock_id  Trading_Volume  ...   open    max    min  close  spread
# 0  2025-06-02     2330        40608468  ...  958.0  961.0  946.0  946.0   -21.0
# 1  2025-06-03     2330        27482916  ...  960.0  965.0  950.0  950.0     4.0

df.dtypes
# date      object   ← 注意：日期是字串！入庫時的型別問題（第 5 章 dtype 講過的「用猜的」）
# close    float64 ...

df.isna().sum()              # ③ 檢查各欄位的空值數量——每一欄都是 0
# 這批資料沒有空值；「檢查」仍是必要步驟。若有空值，用 fillna 填補或 dropna 刪除

接回主線

任務程式 tasks_crawler_finmind_duplicate.py 做的就是上面這三步：打 API 拿 list of dict → 轉成 DataFrame → 逐列 upsert 寫入（下一節逐行解讀）。整條主線是：拿資料（02）→ 轉 DataFrame、檢查（本章前半）→ 冪等入庫（本章後半）。

---

一行一行讀懂「去重寫入」

第一步：定義資料表結構，指定複合主鍵

from sqlalchemy import BigInteger, Column, Date, Float, MetaData, String, Table

metadata = MetaData()
stock_price_table = Table(
    "TaiwanStockPrice_duplicate",                      # 資料表名稱
    metadata,
    Column("stock_id", String(50), primary_key=True), # 主鍵之一
    Column("date", Date, primary_key=True),           # 主鍵之二
    Column("Trading_Volume", BigInteger),
    Column("open", Float),
    Column("max", Float),
    Column("min", Float),
    Column("close", Float),
    # ...其餘欄位
)
metadata.create_all(engine)   # 表不存在才建立

最關鍵的一句：stock_id 和 date 兩個欄位都設了 primary_key=True，這叫複合主鍵。意思是「同一支股票、同一天」在這張表裡只能有一筆——這就是每筆資料的「唯一身分」。上一章的表沒有主鍵，所以資料庫根本不知道哪兩筆算「重複」；這一章明確告訴它了。

第二步：upsert——重複就更新，不重複就新增

from sqlalchemy.dialects.mysql import insert   # MySQL 專用的 insert，支援 on_duplicate_key_update

for _, row in df.iterrows():
    # ① 準備插入這一列
    insert_stmt = insert(stock_price_table).values(**row.to_dict())

    # ② 如果主鍵撞到已存在的資料，就改成「更新所有欄位為新值」
    update_stmt = insert_stmt.on_duplicate_key_update(
        **{col.name: insert_stmt.inserted[col.name]
           for col in stock_price_table.columns}
    )

    # ③ 執行
    with engine.begin() as conn:
        conn.execute(update_stmt)

一段一段看：

• ① insert(...).values(...)：準備把這一列插入表裡。row.to_dict() 把整列轉成 {欄位名: 值} 的 dict（例如 {"stock_id": "2330", "date": "2024-01-02", "close": 593.0, ...}），** 再把這個 dict 展開成 values() 的參數。
• ② .on_duplicate_key_update(...)：這是MySQL 專用語法，也是整章的核心。它的意思是——「如果這筆的主鍵（stock_id + date）跟已存在的撞到了，別報錯、別製造重複，改成把那些欄位更新成新值」。那個 dict comprehension 就是在說「所有欄位都更新成新值」。
• ③ with engine.begin() as conn：在一個交易（transaction）裡執行，確保寫入的完整性。

所以無論這支任務跑幾次，TaiwanStockPrice_duplicate 裡「同股同日」永遠只有一筆，而且是最新的值。這就是冪等。

對照上一章：那邊用 to_sql(if_exists="append")，沒有主鍵概念，只會一直往後疊；這一章用明確的複合主鍵 + upsert，才真正做到冪等。

另一種做法：把去重搬到 DataFrame 層

上面是「DB 層去重」：主鍵擋重複、upsert 撞到改更新。同一件事也可以完全在 pandas 裡做——把庫裡現有的資料讀回來，跟新資料合併、用主鍵欄位去重，再整批寫回：

def df_upsert(new_df):
    current = pd.read_sql("SELECT * FROM TaiwanStockPrice_dfdemo", engine)   # ① 讀回現有資料
    merged = pd.concat([current, new_df], ignore_index=True)                 # ② 新舊合併
    merged["stock_id"] = merged["stock_id"].astype(str)                      # ③ 統一主鍵欄位型別
    merged["date"] = pd.to_datetime(merged["date"])
    merged = merged.drop_duplicates(subset=["stock_id", "date"], keep="last")  # ④ 去重：同股同日留最新
    with engine.begin() as conn:
        conn.execute(text("DELETE FROM TaiwanStockPrice_dfdemo"))            # ⑤ 清空表（結構和主鍵保留）
        merged.to_sql("TaiwanStockPrice_dfdemo", conn, if_exists="append", index=False)  # ⑥ 整批寫回

（可跑版本在 example/pandas.ipynb 最後一段，含建表與驗證：連跑兩輪筆數不變；改一筆值重跑，筆數不變、值被更新。）

drop_duplicates(subset=[...], keep="last") 就是「去重」的 DataFrame 版本：subset 指定「怎樣算同一筆」——跟複合主鍵是同一組欄位；keep="last" 表示重複時留最新的——對應 upsert 的「更新成新值」。

一個必踩的坑在 ③：去重前要先統一型別。current 從 VARCHAR 欄讀回來是字串 "2330"，new_df 若是整數 2330，drop_duplicates 會把它們視為不同的兩筆——去重不會發生，寫回時撞主鍵報 1062。

兩版怎麼選：

	DB 層（本章主線）	DataFrame 層（本節）
去重者	主鍵 + on_duplicate_key_update	drop_duplicates
寫入量	只動有變化的列（增量）	整表清空重寫
多 worker 同時寫	安全（資料庫裁決）	有風險（讀與寫之間，別人可能寫入）
適合	正式排程、資料持續長大	小資料、一次性整理、教學

那為什麼任務程式不直接用 df 版？ 三個原因，都跟我們系統的形態有關：

1. 多 worker 同時寫會互相蓋掉。df 版是「讀回 → 合併 → 清空 → 寫回」四個步驟，不是一個原子操作。兩個 worker 同時執行時：A 讀回資料後，B 也讀回（此時還沒有 A 的新資料），接著 A 清空寫回、B 再清空寫回——B 的結果會把 A 剛寫入的資料整批蓋掉。我們的系統有 worker_twse、worker_tpex 多個 worker 同時在跑，這個風險是真實的。DB 層 upsert 沒有這個問題：每一列的「插入或更新」由資料庫裁決，多個 worker 同時寫互不干擾。
2. 成本隨表長大。df 版每次執行都要把整張表讀回記憶體、再整表重寫；排程每天跑、資料持續累積，這個成本會一直漲。upsert 只處理「這次抓回來的資料」，表多大都一樣。
3. 規則放在對的層。「同股同日只能有一筆」是資料本身的規則。交給資料庫（主鍵）強制執行，任何程式來寫都會被管到；放在 pandas 裡，只有記得呼叫 drop_duplicates 的程式受保護——這正是補充C「約束」一節的觀念。

df 版的定位：單機、一次性的資料整理（例如手動清一批歷史檔案），或像本節這樣把「去重」的邏輯攤開來理解。放進多 worker 的排程系統，就用 DB 層的 upsert。

---

一步一步跟著做：走一遍「append → 報錯 → upsert」

這一輪的流程：先讓 append 跑起來（能動，但重複）→ 加上主鍵（append 立刻不行）→ 引入 upsert（解法）。你會完整走過寫入方式的演化過程。

Step 1：基礎設施

docker compose -f docker-compose-local.yml up -d rabbitmq flower mysql phpmyadmin

另開一個終端機，啟動一個本機 worker——不帶 -Q，收預設佇列。本章不談分流，一個 worker 就夠：

uv run celery -A crawler.worker worker --loglevel=info

Step 2：手動建一張「沒有主鍵」的實驗表

在 phpMyAdmin 的 SQL 頁籤執行：

CREATE TABLE TaiwanStockPrice_duplicate (
  date DATE, stock_id VARCHAR(50),
  Trading_Volume BIGINT, Trading_money BIGINT,
  open FLOAT, max FLOAT, min FLOAT, close FLOAT,
  spread FLOAT, Trading_turnover BIGINT
);      -- 注意：沒有 PRIMARY KEY

然後把 append 版任務指到這張表：打開 crawler/tasks_crawler_finmind.py，找到 df.to_sql("TaiwanStockPrice", ...) 那一行，把表名改成 "TaiwanStockPrice_duplicate"。

改完重啟 worker（Ctrl+C 再重新啟動）——任務程式碼是 worker 啟動時載入的，改了檔案不重啟不會生效。

Step 3：第一輪 append —— 能跑

uv run crawler/producer_crawler_finmind.py

等 worker 做完。✅ append 對「沒主鍵的表」可以成功寫入。

Step 4：第二輪 append —— 還是能跑，但翻倍了

再跑一次同樣的 producer 。append 能跑，代價是重複——沒有主鍵，資料庫不會阻擋重複。

Step 5：刪掉重建「有主鍵」的版本

這是實驗用的表，直接刪掉重建（這次加上複合主鍵）：

DROP TABLE TaiwanStockPrice_duplicate;
CREATE TABLE TaiwanStockPrice_duplicate (
  date DATE NOT NULL, stock_id VARCHAR(50) NOT NULL,
  Trading_Volume BIGINT, Trading_money BIGINT,
  open FLOAT, max FLOAT, min FLOAT, close FLOAT,
  spread FLOAT, Trading_turnover BIGINT,
  PRIMARY KEY (stock_id, date)          -- 這次有主鍵擋重複
);

Step 6：同樣的 append，再跑兩輪

• 第一輪：✅ 成功（表是空的，全部寫入成功）——筆數跟 Step 3 一樣
• 第二輪：❌ 報錯

IntegrityError (1062, "Duplicate entry '2330-2024-01-02' for key 'PRIMARY'")

筆數不變。同一支程式、同一份資料，唯一的差別是表有沒有主鍵：沒主鍵時重複累積、有主鍵時直接報錯——append 不適用。

Step 7：換用 upsert

先把 Step 2 改的表名改回 "TaiwanStockPrice"、重啟 worker。然後跑 upsert 版（它本來就寫 TaiwanStockPrice_duplicate，且遇到既有的主鍵表會直接沿用）：

uv run crawler/producer_crawler_finmind_duplicate_single.py

upsert 撞到主鍵時改成更新，所以：append 在沒主鍵的表會重複、在有主鍵的表會報錯；upsert 在有主鍵的表上重跑任意次，結果都正確。

延伸實驗：把日期範圍加長。 把任務的 start_date 改早（例如 2023-06-01）再跑一次——只有新日期區間的資料會新增，已存在的日期只被更新、不會重複；同一區間再跑一次，筆數維持不變。這就是 upsert 在日常排程的行為：每天執行時，只有新的交易日會新增，其餘都是更新。

檢查你是不是真的做到了

#	你應該看到	它證明了什麼
1	出現 TaiwanStockPrice_duplicate 表	複合主鍵的表建立成功
2	重跑後筆數不變	upsert 讓操作冪等
3	對照 TaiwanStockPrice 會翻倍	有沒有主鍵 + upsert 差很多

---

反向驗證：有主鍵的表，遇上 append 會怎樣？

本章把表加上了複合主鍵、寫入換成 upsert。那如果只做一半——表有主鍵、但寫入還是用第 5 章的 append 版（crawler_finmind）呢？

	表沒有主鍵（第 5 章）	表有主鍵（本章起）
append（普通 INSERT）	✅ 永遠成功——但重跑翻倍（沒有約束阻擋）	第一輪 ✅；第二輪 ❌ 撞 1062 任務失敗
upsert（ON DUPLICATE KEY UPDATE）	✅ 成功，但備案永遠不觸發（形同 append）	✅ 撞主鍵→改更新，兩種情況都正確

為什麼第二輪必定失敗：append 版任務每次執行都「從 2024-01-01 抓整段歷史寫入一遍」，第二輪寫入的資料大部分表裡已經有了——主鍵拒收、報 ERROR 1062、整個任務失敗（錯誤訊息見補充C 分區節）。

三個容易混淆的點，一次釐清：

• 這跟分區無關：普通表加上主鍵，效果一模一樣。補充C 的分區實驗會踩到它，只因 MySQL 規定「分區鍵必須在主鍵裡」——分區逼你裝主鍵；問題出在主鍵，不在分區。
• 這跟 Python 或 SQL 無關：主鍵只看「進來的資料」，不看誰發的 INSERT。用 SQL 搬舊資料撞 1062，是因為舊表內部本來就有重複（append 塞出來的），搬家時自己撞自己——所以搬遷用 INSERT IGNORE 跳過。
• 成對原則：crawler_finmind（append）只能配「沒主鍵的表」；表一旦有主鍵——不管是為了防重複還是為了分區——寫入端就必須換 crawler_finmind_duplicate（upsert）。表的設計和寫入的方式，永遠成對出現。

---

想再深入一點

• 為什麼這一章要自己用 Table(...) 定義結構，第 5 章卻不用？ 因為 on_duplicate_key_update（upsert）要能運作，資料庫必須先知道「哪些欄位是主鍵」——它是靠「主鍵撞到了沒」來判斷該 insert 還是 update。第 5 章的 to_sql 自動建表不會幫你設主鍵，所以這一章改成明確用 Table(...) 宣告 stock_id + date 為複合主鍵，upsert 才有依據。
• insert_stmt.inserted[col.name] 是什麼？ 這是 MySQL upsert 的慣用寫法，inserted[...] 代表「這次原本要插入的那個新值」。所以 on_duplicate_key_update 那段的意思就是「如果撞到主鍵，把每個欄位都更新成這次帶進來的新值」——等於用新資料覆蓋舊資料。
• 為什麼用 iterrows() 一列一列寫？慢不慢？ 慢。iterrows() 逐列跑迴圈、一列送一次 SQL，資料量大時效率不好。教學這樣寫是為了讓你看清楚「每一列都在做 upsert」的邏輯；正式環境會改成「批次 upsert」（一次送多列）來加速。你之後優化時可以往這個方向想。
• 換成別的資料庫呢？ on_duplicate_key_update 是 MySQL 專屬語法（所以 import 的是 sqlalchemy.dialects.mysql.insert）。如果換成 PostgreSQL，要改用 ON CONFLICT (...) DO UPDATE；概念一樣（upsert），但語法不同。這也是為什麼 import 要特別指定 MySQL 方言。
• 主鍵怎麼選是門學問。 這裡選 stock_id + date 是因為它能唯一辨識「某股某日」這筆商業資料。換一種資料就要重想：例如「使用者的每日簽到」可能是 user_id + date、「訂單明細」可能是 order_id + product_id。選錯主鍵，冪等就會失效（要嘛擋不掉重複，要嘛蓋掉不該蓋的資料）。

---

想一想（確認你懂了）

先自己想過再看答案。

Q1：用一句話解釋「冪等」，並舉一個生活例子。

冪等 = 同一個操作做一次跟做很多次，結果一樣。生活例子：電梯樓層鍵（連按不會多帶你去幾層）、或「把開關撥到 ON」（撥幾次都還是 ON）。相對地，「再倒一杯水」就不冪等（做幾次就多幾杯）。

Q2：為什麼 stock_id 單獨當主鍵不夠？為什麼要加 date？

因為同一支股票有很多天的資料，如果只用 stock_id 當主鍵，2330 就只能存一筆，第二天的資料會把第一天蓋掉。要能「一支股票、多天各一筆」，唯一身分必須是「股票 + 日期」的組合，所以要用 stock_id + date 複合主鍵。

Q3：如果 FinMind 事後修正了某天的收盤價，重跑這支任務會發生什麼？

因為主鍵（該股該日）已存在，upsert 會走「更新」那條路，把那一筆的欄位更新成新的收盤價，而不是新增一筆。所以你的資料會被自動修正成最新值，且不會產生重複。這正是冪等在真實維運裡的價值。

---

換你試試看

練習 1：對照「冪等」vs「不冪等」

先跑第 5 章的 producer_crawler_finmind.py 三次，看 TaiwanStockPrice 筆數一直漲；再跑這章的 producer_crawler_finmind_duplicate_single.py 三次，看 TaiwanStockPrice_duplicate 筆數不動。把兩組數字並排寫下來，你就把「冪等」這個抽象詞變成看得到的東西了。

練習 2：驗證「重複真的被去掉了」

對去重表跑第 5 章練習 2 的那條 SQL：

SELECT stock_id, date, COUNT(*)
FROM TaiwanStockPrice_duplicate
GROUP BY stock_id, date
HAVING COUNT(*) > 1;

你應該得到空結果（沒有任何一組重複）。對照第 5 章的表會查到一堆重複。這讓你確認複合主鍵 + upsert 真的生效。

練習 3：手動改一筆再重跑，觀察 update

在 phpMyAdmin 手動把去重表裡某支股票某天的 close 改成一個明顯錯的值（例如 999999），然後重跑 producer。回去看那一筆——它會被改回 FinMind 的正確值。這顯示 upsert 的「update」那一半在運作：重複的主鍵不是被跳過，是被更新。

---

卡住了？常見錯誤這樣排

你遇到的狀況	原因	怎麼解
on_duplicate_key_update 語法錯	這是 MySQL 專用語法	確認 import 的是 sqlalchemy.dialects.mysql.insert；換 PostgreSQL 要改用 ON CONFLICT
重跑筆數還是變多	主鍵沒設對，或寫到了沒有主鍵的表	確認寫入的是 TaiwanStockPrice_duplicate、且 stock_id + date 都是主鍵
逐列 upsert 覺得慢	iterrows() 一列一列寫	教學夠用；正式環境可改批次 upsert

---

這一章你學到了

• 冪等 = 同操作做幾次結果都一致，是定期資料流的必備性質。
• 複合主鍵定義「一筆資料的唯一身分」，upsert 讓重跑安全。
• MySQL 用 on_duplicate_key_update 實作 upsert。

下一章要做什麼

到這裡，抓取、可靠性、落地、冪等都齊了。下一章把你一路用到的三個 Web 管理介面（RabbitMQ、Flower、phpMyAdmin）做個總整理，建立一套「出事了先看哪裡」的排錯 SOP，為 Celery 階段收尾。