第 2 章：資料蒐集與清理

# 第 2 章：資料蒐集與清理 在數據決策的生命週期中，資料蒐集與清理是決定後續分析品質與商業洞察可信度的關鍵環節。本章將從資料來源的分類、蒐集方法談起，深入探討缺失值處理、重複資料剔除、離群值偵測、資料標準化與正規化等清理技巧，並以實務範例說明如何將雜亂的原始資料轉化為可供模型訓練與決策使用的乾淨資料集。 --- ## 1. 資料來源分類 | 類別 | 來源範例 | 資料類型 | 典型挑戰 | |------|----------|----------|----------| | 內部 | ERP、CRM、交易系統 | 結構化（SQL） | 權限控制、資料庫版本差異 | | 內部 | 內部日誌、APP 使用紀錄 | 半結構化/非結構化 | 格式多樣、欄位不統一 | | 外部 | API（天氣、金融、社群） | 結構化 | 認證、速率限制 | | 外部 | 網路爬蟲（新聞、評論） | 非結構化 | 版權、反爬機制 | | 外部 | 開放資料集（政府、科研） | 結構化 | 版本更新頻繁 | > **實務提醒**：在決定資料來源時，務必先完成「資料需求清單」與「合規性檢查」兩項工作，確保未來蒐集的資料不會因版權或隱私問題產生法律風險。 --- ## 2. 資料蒐集流程 1. **需求定義**：將商業問題拆解為「需要哪類資料」與「資料更新頻率」兩個維度。 2. **來源確認**：評估來源可靠度、數據質量、授權範圍。 3. **接入方式**： - SQL: `SELECT`、`JOIN`、資料抽取腳本。 - REST API: `GET`、`POST`、Token 認證。 - 資料倉儲（Data Lake）：使用 `AWS S3` / `Azure Blob` 直接存取。 4. **自動化**：使用 Airflow、Luigi 或 Prefect 編排 ETL 週期。 5. **監控與告警**：監測失敗率、延遲、資料完整性。 python # Airflow DAG 範例：每日從 API 取得天氣資料並寫入資料庫 from airflow import DAG from airflow.operators.python import PythonOperator from datetime import datetime, timedelta import requests, json, psycopg2 default_args = { 'owner': 'data_engineer', 'depends_on_past': False, 'retries': 1, 'retry_delay': timedelta(minutes=5), } def fetch_weather(**context): url = 'https://api.weather.gov/gridpoints/MDL/96,68/forecast' r = requests.get(url, timeout=10) r.raise_for_status() data = r.json() # 這裡簡化，只提取溫度與天氣描述 rows = [] for period in data['properties']['periods']: rows.append((period['name'], period['temperature'], period['shortForecast'])) # 寫入 PostgreSQL conn = psycopg2.connect("dbname=weather user=postgres") cur = conn.cursor() cur.executemany("INSERT INTO daily_forecast (name, temperature, description) VALUES (%s,%s,%s)", rows) conn.commit() cur.close(); conn.close() with DAG('weather_etl', default_args=default_args, schedule_interval='@daily', start_date=datetime(2023, 1, 1)) as dag: t1 = PythonOperator(task_id='fetch_weather', python_callable=fetch_weather) --- ## 3. 資料清理技巧 ### 3.1 缺失值處理 | 方法 | 適用情境 | 優點 | 缺點 | |------|----------|------|------| | 刪除 | 缺失比例 <5% 且隨機缺失 | 簡單、快速 | 可能喪失資訊 | | 取代（均值/中位數/眾數） | 數值型、離散型資料 | 維持樣本數 | 可能低估變異性 | | 預測填補 | 缺失比例高、非隨機缺失 | 可保留更多樣本 | 需要建立填補模型 | | 多重插補 | 需要統計推論 | 考慮不確定性 | 複雜、耗時 | > **實務範例**：在客戶資料庫中，`email` 欄位缺失率 2%，可直接刪除；而 `age` 缺失率 15%，建議使用多重插補或預測填補。 ### 3.2 重複資料剔除 python import pandas as pd df = pd.read_csv('raw_data.csv') # 檢查重複 print('重複列數:', df.duplicated().sum()) # 刪除重複 cleaned = df.drop_duplicates() > **提示**：在分布式系統（Spark、Dask）下，可使用 `dropDuplicates()` 以避免資料分片之間的重複。 ### 3.3 離群值偵測 | 方法 | 公式 | 應用 | |------|------|------| | IQR | Q3 – Q1 | 連續數值 | | 標準差 | μ ± 3σ | 連續數值 | | Z-score | (x-μ)/σ | 連續數值 | | DBSCAN | 距離 | 空間/非線性 | python from scipy import stats import numpy as np # 標準差法 z_scores = np.abs(stats.zscore(df['purchase_amount'])) df = df[(z_scores < 3)] ### 3.4 資料標準化與正規化 | 技術 | 公式 | 何時使用 | |------|------|----------| | Z-Score | (x-μ)/σ | 需要距離度量的模型（KNN、SVM） | | Min-Max | (x-min)/(max-min) | 神經網路前處理 | | Log Transform | log(x+1) | 右偏資料 | > **實務範例**：對銷售金額進行 Log 轉換後，再做 Z-Score 標準化，可降低極端值對模型的影響。 --- ## 4. 資料質量指標 (Data Quality Metrics) | 指標 | 定義 | 評估方式 | |------|------|----------| | 完整性 | 所有必要欄位是否皆有資料 | 欄位缺失率 | | 正確性 | 資料符合業務邏輯 | 例：年齡 > 0、郵遞區號長度 | | 一致性 | 跨資料表或資料源資料不矛盾 | 參照鍵檢查 | | 時效性 | 資料更新頻率 | 資料最後更新時間 | | 唯一性 | 同一實體不重複 | Primary key 檢查 | > **監控工具**：使用 Data Quality Dashboard（如 Great Expectations）自動化測試。 --- ## 5. 案例研究：零售商客戶資料清理流程 | 步驟 | 目標 | 工具 | 重點 | |------|------|------|------| | 1 | 來源整合 | Airflow + Snowflake | 多個系統資料抽取、合併 | | 2 | 缺失處理 | Pandas + SQL | 先刪除 1% 的缺失，後預測 15% 缺失 | | 3 | 重複剔除 | Dask | 大規模資料集 | | 4 | 離群值檢測 | Prophet | 促銷期間銷售異常 | | 5 | 正規化 | scikit-learn | 分類模型前處理 | | 6 | 資料驗證 | Great Expectations | 生成報告 | > **成果**：資料品質提升 12%，模型預測準確率提升 3%。 --- ## 6. 小結 1. 資料蒐集前須完成需求與合規性確認，避免未來法律風險。 2. 清理流程中，缺失值、重複值、離群值與標準化是三大核心步驟。 3. 應用合適的工具（Pandas、Spark、Great Expectations）與自動化流程，能大幅提升效率與資料可靠性。 4. 資料品質指標不僅是「檢查」，更是持續監控的關鍵。 > **後記**：下一章將進一步探討如何利用清理後的資料進行探索性資料分析，發掘商業潛在洞察，為決策提供更堅實的數據支撐。