返回目錄
A
數據驅動決策：從原始資料到洞察的全流程 - 第 2 章
第 2 章：資料蒐集與獲取
發布於 2026-02-22 13:38
# 第 2 章：資料蒐集與獲取
資料是數據科學的原料，資料蒐集與獲取階段決定了後續分析的可行性與品質。 本章將帶你從概念到實務，了解
1. **資料來源分類**：結構化、非結構化、實時流；
2. **獲取手段**：API、網路爬蟲、企業內部資料庫、感測器、第三方資料供應商；
3. **工具與技術**：SQL、Pandas、Spark、Kafka、REST/GraphQL、Scrapy、Selenium 等；
4. **合規與治理**：授權、隱私、資料治理。
---
## 2.1 資料來源概覽
| 資料類型 | 特徵 | 優勢 | 典型範例 |
|----------|------|------|----------|
| **結構化資料** | 行（rows）×列（columns） | 方便 SQL 查詢、數據完整性高 | 交易記錄、客戶資料、會計帳簿 |
| **半結構化資料** | 如 JSON、XML | 具備靈活性、可擴充 | 網站 API 回應、物聯網感測數據 |
| **非結構化資料** | 文字、影像、音訊、影片 | 大量資訊、語意豐富 | 產品評論、社群貼文、客服錄音 |
| **實時流資料** | 持續進入的事件序列 | 低延遲決策、即時分析 | 金融交易、網路行為、IoT 事件 |
> **實務洞察**：在建模前先做「資料來源盤點」表（See Table 2‑1）能明確哪些資料可直接使用、哪些需要預處理或合併。
| 資料來源 | 是否可即時使用 | 需要轉換 | 典型工具 |
|----------|--------------|----------|-----------|
| 企業內部資料庫 | ✔ | 否 | MySQL, PostgreSQL, SQL Server, Snowflake |
| 第三方 API | ✔ | 需要 | Requests, HTTPX, GraphQL client |
| Web 站點 | ⚠ | 高 | Scrapy, BeautifulSoup, Selenium |
| 物聯網感測 | ✔ | 需要 | MQTT, Kafka, Spark Structured Streaming |
| 社群媒體 | ⚠ | 高 | Tweepy, Facebook Graph API, Instagram API |
## 2.2 資料獲取流程圖
mermaid
flowchart TD
A[需求定義] --> B[資料來源分析]
B --> C{來源類型}
C -->|結構化| D[SQL 連線]
C -->|半結構化| E[API 請求]
C -->|非結構化| F[爬蟲]
C -->|實時流| G[Kafka / Spark]
D --> H[資料檢索]
E --> H
F --> H
G --> H
H --> I[資料存儲]
I --> J[資料治理]
## 2.3 具體實作示例
### 2.3.1 連接企業資料庫（PostgreSQL）
python
import psycopg2
import pandas as pd
conn = psycopg2.connect(
dbname='sales_db',
user='db_user',
password='secret',
host='10.0.0.5',
port=5432
)
query = """
SELECT customer_id, purchase_date, amount
FROM transactions
WHERE purchase_date >= '2023-01-01';
"""
df = pd.read_sql(query, conn)
print(df.head())
> **提示**：使用 `sqlalchemy` 可輕鬆轉換為 ORM，提升可維護性。
### 2.3.2 呼叫 REST API（Python Requests）
python
import requests
import pandas as pd
url = 'https://api.example.com/v1/customers'
headers = {
'Authorization': 'Bearer YOUR_ACCESS_TOKEN',
'Accept': 'application/json'
}
response = requests.get(url, headers=headers, params={'page': 1, 'size': 100})
if response.status_code == 200:
data = response.json()
df = pd.DataFrame(data['customers'])
print(df.info())
else:
print('Error:', response.status_code)
### 2.3.3 網路爬蟲（Scrapy 範例）
python
# 在 scrapy project 裡新增 spiders/real_estate_spider.py
import scrapy
class RealEstateSpider(scrapy.Spider):
name = 'real_estate'
start_urls = ['https://example.com/house-list']
def parse(self, response):
for house in response.css('.house-card'):
yield {
'title': house.css('.title::text').get(),
'price': house.css('.price::text').get(),
'url': house.css('a::attr(href)').get(),
}
> **安全注意**：遵守 `robots.txt` 規則，且若需要瀏覽器互動可用 Selenium，但需留意 Cookie & User‑Agent 篡改。
### 2.3.4 實時流（Kafka + Spark Structured Streaming）
python
from pyspark.sql import SparkSession
spark = SparkSession.builder \
.appName('KafkaStreaming') \
.getOrCreate()
stream_df = spark.readStream \
.format('kafka') \
.option('kafka.bootstrap.servers', 'broker1:9092,broker2:9092') \
.option('subscribe', 'user_events') \
.load()
# 解析 key/value JSON
stream_df = stream_df.selectExpr("CAST(value AS STRING) as json_str") \
.selectExpr("json_tuple(json_str, 'user_id', 'action', 'timestamp') as (user_id, action, ts)")
query = stream_df.writeStream \
.format('delta') \
.outputMode('append') \
.option('checkpointLocation', '/checkpoints/user_events') \
.start('/data/warehouse/user_events')
query.awaitTermination()
## 2.4 資料倉儲策略
| 儲存方案 | 主要用途 | 優勢 | 典型技術 |
|----------|----------|------|-----------|
| **資料倉庫 (DW)** | 事務性、結構化查詢 | ACID、易於 BI | Snowflake, BigQuery, Redshift |
| **資料湖 (DL)** | 大量原始資料、非結構化 | 彈性、可延伸 | S3, HDFS, Azure Data Lake |
| **資料湖倉 (DLC)** | 結構化 + 非結構化 | 高效、易於探索 | Databricks, Hive, Glue |
> **合規**：在將資料寫入資料湖時，務必先加入 *metadata schema*（例如 `Delta Lake` 的 schema enforcement），確保
> 資料類型與結構一致。
## 2.5 合規與治理
| 規範 | 主要關注點 | 具體措施 |
|------|-------------|-----------|
| **GDPR / CCPA** | 個人資料、同意 | 需求同意收集、刪除權、資料匿名化 |
| **資料授權** | 版權、商業授權 | 取得授權文件、使用 `pydantic` 進行 schema 驗證 |
| **資料品質** | 完整性、正確性 | 使用 `Great Expectations` 或自訂 schema validation |
| **資料血統** | Provenance、版本管理 | 加入 `DataHub`、`Amundsen` 等資料目錄 |
> **實務案例**：在蒐集社群貼文時，必須檢查 API 供應商的使用條款，若有「商業用途限制」，需重新談判或尋找備援來源。
## 2.6 小結
1. **先定義需求**：清楚知道要做什麼，才能篩選合適的資料來源。
2. **分類資料**：結構化資料最容易即時使用，非結構化資料需要更複雜的前處理。
3. **選擇技術**：根據來源類型選擇適合的工具，例如 SQL 連線、REST API、爬蟲、Kafka。
4. **確保合規**：授權、隱私與治理不是後台工作，而是資料蒐集的基礎。
> **最佳實踐**：建立「資料蒐集工作流程」與「資料源評估表」兩份文件，作為資料工程師與業務部門溝通的共用語言。
---
> **下一章預告**：在第 3 章「資料探索與清洗」中，我們將進一步探討如何驗證資料品質、處理缺失值與重複項，將本章蒐集到的原始資料轉化為可用的分析資料集。