Pandas 資料清理實戰：缺失值、型別、重複資料與驗證

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

§ 117: 本文

資料讀得進來，不代表資料已經能用。

空字串、N/A、None 可能同時代表缺值；訂單數量看起來是數字，實際 dtype 卻是字串；同一筆訂單更新兩次，又可能被當成兩筆營收。這些問題不一定立刻噴錯，反而會安靜地污染後面的分析。

這篇拍拍君會把資料清理整理成一條可重跑、可驗證、可稽核的 Pandas workflow：

先盤點，不急著修改；
統一欄名與缺值語意；
明確解析每個欄位的 dtype；
按業務鍵處理重複資料；
用資料契約擋住不合理資料；
保留清理摘要，再輸出結果。

如果你想做的是 GroupBy、Pivot Table 與報表，先看這篇 Pandas GroupBy 實戰；本文不重複聚合，而是處理它前面的「資料能不能信」。

一. 安裝與建立專案
#

用 uv 建一個最小專案：

uv init pandas-cleaning-demo
cd pandas-cleaning-demo
uv add "pandas>=3.0,<4"

本文只依賴 Pandas。資料契約會先用普通 Python 函式實作，這樣你能看清楚每一條規則，而不是把問題藏進另一個框架。

建立 clean_orders.py：

from __future__ import annotations
from dataclasses import dataclass
from io import StringIO
import pandas as pd

先準備一份刻意弄髒的訂單資料：

RAW_CSV = """order_id, customer ,quantity,unit_price,status,ordered_at,updated_at
 A-001 ,拍拍醬,2,120.5,paid,2026-09-01,2026-09-01 09:10
A-002,chatPTT,,80,PAID,2026/09/02,2026-09-02 10:00
A-002,chatPTT,3,80,paid,2026/09/02,2026-09-02 10:30
A-003,,1,unknown,cancelled,2026-09-03,2026-09-03 11:00
A-004,拍拍君,-2,50,pending,bad-date,2026-09-04 12:00
A-005,N/A,4,30,shipped,2026-09-05,2026-09-05 13:00
"""
orders = pd.read_csv(StringIO(RAW_CSV), keep_default_na=True)

這裡的坑包括欄名空白、大小寫不一致、缺少數量、壞掉的價格與日期、負數，以及重複的 order_id。很像故意考人，對吧？現實資料通常更壞。

二. 第一件事是盤點，不是 `dropna()`
#

一看到缺值就整列刪掉，是最快也最危險的清理方式。先看 shape、dtype、缺值數量與樣本：

def profile(df: pd.DataFrame) -> pd.DataFrame:
    return pd.DataFrame(
        {
            "dtype": df.dtypes.astype("string"),
            "missing": df.isna().sum(),
            "missing_rate": df.isna().mean().round(3),
            "unique": df.nunique(dropna=False),
        }
    )

print(orders.shape)
print(profile(orders))
print(orders.head())

isna() 可以辨認 Pandas 已知的缺值 sentinel，但不會自動知道 ""、"unknown" 或 "-" 在你的業務裡是不是缺值。那是資料契約的一部分，必須由你定義。

盤點時至少問四件事：

應該唯一的 key 是否重複？
數值欄為什麼被推斷成字串？
缺值集中在哪些欄位？
類別值是否只是大小寫或空白不同？

三. 先整理欄名與字串
#

欄名是資料契約的入口。拍拍君通常先轉成小寫 snake_case，並拒絕整理後撞名的欄位：

def normalize_columns(df: pd.DataFrame) -> pd.DataFrame:
    renamed = (
        df.columns.str.strip()
        .str.lower()
        .str.replace(r"[^0-9a-zA-Z]+", "_", regex=True)
        .str.strip("_")
    )
    duplicated = renamed[renamed.duplicated()].tolist()
    if duplicated:
        raise ValueError(f"欄名整理後重複：{duplicated}")
    return df.set_axis(renamed, axis="columns").copy()

接著只清理真正的文字欄，不要對整張表盲目 astype(str)：

def normalize_text(df: pd.DataFrame) -> pd.DataFrame:
    result = df.copy()
    for column in ["order_id", "customer", "status"]:
        result[column] = result[column].astype("string").str.strip()
    result["order_id"] = result["order_id"].str.upper()
    result["status"] = result["status"].str.lower()
    custom_missing = {"": pd.NA, "n/a": pd.NA, "unknown": pd.NA, "-": pd.NA}
    result["customer"] = result["customer"].str.lower().replace(custom_missing)
    return result

為什麼不使用 astype(str)？因為它可能把真正的缺值變成字串 "nan" 或 "<NA>"。後面再呼叫 isna()，這些字串已經不算缺值了。

四. 缺失值要按欄位角色處理
#

不是所有缺值都該填，也不是所有缺值都該刪。可以先替欄位分類：

欄位角色	例子	常見策略
主鍵	`order_id`	缺值直接拒絕
必要度量	`quantity`	規則允許才填，否則隔離
選填文字	`customer`	保留 `pd.NA` 或填明確標籤
時間	`ordered_at`	解析失敗要記錄，不能亂補今天
類別	`status`	映射、驗證允許值

先把多種缺值表示統一成 nullable dtype：

def use_nullable_dtypes(df: pd.DataFrame) -> pd.DataFrame:
    return df.convert_dtypes(dtype_backend="numpy_nullable")

nullable integer 的名稱是大寫 Int64，可以同時保存整數與 pd.NA。這比因為一個缺值就把數量欄變成 float64 更符合語意。

填值要有理由。例如「缺少數量代表一件」只有在來源系統明確保證時才成立：

result["quantity"] = result["quantity"].fillna(1)

若沒有這個保證，保留缺值並讓驗證失敗，通常比安靜猜答案更安全。

五. 用 `to_numeric()` 與 `to_datetime()` 顯式解析
#

轉型失敗最實用的做法，通常是先 errors="coerce"，再把新產生的缺值列抓出來：

def parse_types(df: pd.DataFrame) -> tuple[pd.DataFrame, pd.DataFrame]:
    result = df.copy()
    raw_price = result["unit_price"].copy()
    raw_date = result["ordered_at"].copy()
    result["quantity"] = pd.to_numeric(result["quantity"], errors="coerce").astype("Int64")
    result["unit_price"] = pd.to_numeric(result["unit_price"], errors="coerce").astype("Float64")
    result["ordered_at"] = pd.to_datetime(result["ordered_at"], errors="coerce", format="mixed")
    result["updated_at"] = pd.to_datetime(result["updated_at"], errors="coerce")
    parse_errors = result.loc[
        (raw_price.notna() & result["unit_price"].isna())
        | (raw_date.notna() & result["ordered_at"].isna()),
        ["order_id", "unit_price", "ordered_at"],
    ].copy()
    return result, parse_errors

errors="coerce" 不是把錯誤當作沒看見，而是把十種奇怪輸入壓成一種可偵測狀態。重點是下一步必須檢查 parse_errors。

解析日期時也不要用 fillna(pd.Timestamp.today())。一筆未知日期突然變成今天，通常比 NaT 更難追查。

六. 重複資料不是「整列一樣」才算
#

drop_duplicates() 預設比較所有欄位，但業務上常用主鍵判斷同一筆資料。範例裡 A-002 的兩列 updated_at 不同，整列並不相同，卻代表同一張訂單的兩個版本。

先把所有衝突列列出來：

def find_duplicate_keys(df: pd.DataFrame) -> pd.DataFrame:
    mask = df.duplicated(subset=["order_id"], keep=False)
    return df.loc[mask].sort_values(["order_id", "updated_at"])

duplicates = find_duplicate_keys(orders)

若契約規定「更新時間較晚者勝出」，就明確排序後保留最後一筆：

def keep_latest(df: pd.DataFrame) -> pd.DataFrame:
    return (
        df.sort_values(["order_id", "updated_at"], na_position="first")
        .drop_duplicates(subset=["order_id"], keep="last")
        .reset_index(drop=True)
    )

不要先 drop_duplicates() 再假裝衝突不存在。正式流程應把 duplicates 數量寫進報告，必要時另存隔離檔。

七. 資料契約：把「看起來怪」變成可執行規則
#

我們需要的契約如下：

必要欄位全部存在；
order_id 不缺值且唯一；
quantity 必須大於零；
unit_price 不得為負；
status 只能是 paid、pending、cancelled、shipped；
日期解析必須成功。

先定義驗證結果：

@dataclass(frozen=True)
class ValidationIssue:
    rule: str
    rows: tuple[int, ...]
    detail: str

再把規則集中到一個函式：

REQUIRED_COLUMNS = {
    "order_id",
    "customer",
    "quantity",
    "unit_price",
    "status",
    "ordered_at",
    "updated_at",
}
ALLOWED_STATUS = {"paid", "pending", "cancelled", "shipped"}

def validate_orders(df: pd.DataFrame) -> list[ValidationIssue]:
    missing_columns = REQUIRED_COLUMNS - set(df.columns)
    if missing_columns:
        return [
            ValidationIssue(
                rule="required_columns",
                rows=(),
                detail=f"缺少欄位：{sorted(missing_columns)}",
            )
        ]
    checks = {
        "order_id_required": df["order_id"].isna(),
        "order_id_unique": df["order_id"].duplicated(keep=False),
        "quantity_positive": df["quantity"].isna() | df["quantity"].le(0),
        "unit_price_non_negative": df["unit_price"].isna() | df["unit_price"].lt(0),
        "status_allowed": ~df["status"].isin(ALLOWED_STATUS),
        "ordered_at_valid": df["ordered_at"].isna(),
        "updated_at_valid": df["updated_at"].isna(),
    }
    issues = []
    for rule, failed in checks.items():
        rows = tuple(df.index[failed.fillna(True)].tolist())
        if rows:
            issues.append(
                ValidationIssue(rule=rule, rows=rows, detail=f"失敗 {len(rows)} 列")
            )
    return issues

這個版本故意回傳全部問題，而不是第一個錯誤就 raise。批次資料若一次只能修一條規則，會變成很煩的打地鼠。

八. 組成可重跑的清理管線
#

把順序固定，避免 Notebook 裡手動執行 cell 造成狀態不同：

@dataclass(frozen=True)
class CleaningReport:
    input_rows: int
    output_rows: int
    duplicate_rows: int
    parse_error_rows: int
    validation_issues: tuple[ValidationIssue, ...]

def clean_orders(raw: pd.DataFrame) -> tuple[pd.DataFrame, CleaningReport]:
    cleaned = normalize_columns(raw)
    cleaned = normalize_text(cleaned)
    cleaned = use_nullable_dtypes(cleaned)
    cleaned, parse_errors = parse_types(cleaned)
    duplicate_rows = len(find_duplicate_keys(cleaned))
    cleaned = keep_latest(cleaned)
    issues = tuple(validate_orders(cleaned))
    report = CleaningReport(
        input_rows=len(raw),
        output_rows=len(cleaned),
        duplicate_rows=duplicate_rows,
        parse_error_rows=len(parse_errors),
        validation_issues=issues,
    )
    return cleaned, report

執行：

cleaned, report = clean_orders(orders)
print(cleaned)
print(report)
if report.validation_issues:
    raise ValueError(f"資料驗證失敗：{report.validation_issues}")

這份刻意弄髒的資料會失敗，因為 A-003 價格壞掉、A-004 數量與日期不合法。這是正確結果：清理程式的工作不是讓所有輸入都變成綠燈，而是阻止壞資料混進下游。

九. 隔離錯誤列，不要默默刪除
#

若批次工作不能因少數錯誤整批停止，可以把資料分成 accepted 與 rejected：

def split_valid_rows(df: pd.DataFrame) -> tuple[pd.DataFrame, pd.DataFrame]:
    invalid = (
        df["order_id"].isna()
        | df["quantity"].isna()
        | df["quantity"].le(0)
        | df["unit_price"].isna()
        | df["unit_price"].lt(0)
        | ~df["status"].isin(ALLOWED_STATUS)
        | df["ordered_at"].isna()
        | df["updated_at"].isna()
    ).fillna(True)
    accepted = df.loc[~invalid].reset_index(drop=True)
    rejected = df.loc[invalid].reset_index(drop=True)
    return accepted, rejected

輸出時把兩份資料分開：

accepted, rejected = split_valid_rows(cleaned)
accepted.to_csv("orders_clean.csv", index=False)
rejected.to_csv("orders_rejected.csv", index=False)

隔離檔至少要保留原始 key、失敗欄位與原因。真實系統還應加上 batch ID、來源檔名與處理時間，才能回答「這筆資料為什麼沒進報表？」

十. 合併資料時也要驗證關係
#

清理完主表後，merge() 仍可能因重複 key 把列數乘開。可以用 validate 宣告預期關係：

customers = pd.DataFrame(
    {
        "customer": ["拍拍醬", "chatptt", "拍拍君"],
        "segment": ["A", "B", "A"],
    }
).convert_dtypes()
enriched = accepted.merge(
    customers,
    on="customer",
    how="left",
    validate="many_to_one",
)

如果右表 customer 重複，Pandas 會直接拒絕這次合併。這比合併後才發現營收突然翻倍可靠得多。

十一. 清理後的測試清單
#

最少替穩定規則留幾個 assertions：

def test_clean_orders() -> None:
    cleaned, report = clean_orders(orders)
    assert report.input_rows == 6
    assert report.output_rows == 5
    assert report.duplicate_rows == 2
    assert cleaned["order_id"].is_unique
    assert str(cleaned["quantity"].dtype) == "Int64"
    assert cleaned.loc[cleaned["order_id"].eq("A-002"), "quantity"].item() == 3