ETL入门：数据仓库与数据管道基础

      事实表（Fact Table）
     ┌──────────────────────┐
     │ 订单ID │ 金额 │ 日期 │
     └────┬──────────┬──────┘
          │          │
     ┌────▼──┐  ┌────▼────┐
     │ 用户维度│  │ 产品维度 │
     └────────┘  └─────────┘

抽取 → 转换（在中间层） → 加载到数仓

抽取 → 加载到数仓 → 在数仓内部转换（SQL）

# 伪代码
df = source.read("SELECT * FROM orders")
target.write(df)

# 伪代码 - 基于时间戳
last_max = target.get_max("update_time")
df = source.read(f"SELECT * FROM orders WHERE update_time > '{last_max}'")
target.write(df)

import pandas as pd

df = pd.read_csv("raw_data.csv")

# ① 去除前后空格
df['name'] = df['name'].str.strip()

# ② 处理空值
df['age'] = df['age'].fillna(0)                     # 数值列填 0
df['email'] = df['email'].fillna('unknown@')         # 字符串列填默认值

# ③ 统一格式
df['phone'] = df['phone'].str.replace(r'\D', '', regex=True)  # 去除非数字

# ④ 去除重复行
df = df.drop_duplicates(subset=['user_id'])

# ⑤ 过滤异常值
df = df[df['amount'] >= 0]

# ⑥ 类型转换
df['created_at'] = pd.to_datetime(df['created_at'])

# ⑦ 衍生字段
df['year_month'] = df['created_at'].dt.strftime('%Y-%m')

pip install pandas sqlalchemy pymysql  # MySQL 驱动
# 或者
pip install pandas sqlalchemy duckdb   # 用 DuckDB 做本地测试（无需数据库服务）

"""
etl_demo.py - 一个完整的 ETL 示例
从 SQLite 源库抽取订单数据 → 清洗转换 → 加载到目标库
"""

import pandas as pd
from sqlalchemy import create_engine, MetaData, Table, Column, Integer, String, Float, DateTime, Date
from datetime import datetime, timedelta
import logging

# ========== 配置日志 ==========
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
logger = logging.getLogger(__name__)


# ========== 第一步：Extract（抽取） ==========

def extract(source_conn_str: str, sql: str) -> pd.DataFrame:
    """
    从源数据库抽取数据
    """
    logger.info(f"[Extract] 开始从源库抽取数据...")
    engine = create_engine(source_conn_str)
    with engine.connect() as conn:
        df = pd.read_sql(sql, conn)
    logger.info(f"[Extract] 抽取完成，共 {len(df)} 行")
    return df


# ========== 第二步：Transform（转换/清洗） ==========

def transform(raw_df: pd.DataFrame) -> pd.DataFrame:
    """
    数据清洗与转换
    """
    logger.info(f"[Transform] 开始清洗数据，原始行数: {len(raw_df)}")

    df = raw_df.copy()

    # --- 数据清洗 ---

    # 1. 删除完全重复行
    before = len(df)
    df = df.drop_duplicates()
    logger.info(f"  去重: {before} → {len(df)}（删除 {before - len(df)} 行）")

    # 2. 处理空值
    df['customer_name'] = df['customer_name'].fillna('未知客户')
    df['amount'] = df['amount'].fillna(0.0)
    df['status'] = df['status'].fillna('pending')

    # 3. 去除字符串前后空格
    for col in ['customer_name', 'product', 'status']:
        if col in df.columns:
            df[col] = df[col].astype(str).str.strip()

    # 4. 金额必须 >= 0，负值取绝对值（也可标记为异常）
    df['amount'] = df['amount'].abs()

    # 5. 统一状态字段
    status_mapping = {'待处理': 'pending', '已完成': 'completed', '已取消': 'cancelled'}
    df['status'] = df['status'].map(status_mapping).fillna(df['status'])

    # --- 数据转换 ---

    # 6. 时间字段标准化
    df['order_date'] = pd.to_datetime(df['order_date'], errors='coerce')
    df['order_date_str'] = df['order_date'].dt.strftime('%Y-%m-%d')

    # 7. 衍生字段：订单月份
    df['order_month'] = df['order_date'].dt.to_period('M').astype(str)

    # 8. 计算字段：金额等级
    df['amount_level'] = pd.cut(
        df['amount'],
        bins=[0, 100, 500, float('inf')],
        labels=['小额', '中额', '大额']
    )

    logger.info(f"[Transform] 清洗完成，最终行数: {len(df)}")
    return df


# ========== 第三步：Load（加载） ==========

def load(target_conn_str: str, table_name: str, df: pd.DataFrame, if_exists: str = 'replace'):
    """
    加载数据到目标数据仓库
    if_exists: 'replace'（覆写） / 'append'（追加） / 'upsert'（需自定义）
    """
    logger.info(f"[Load] 开始加载数据到目标表 {table_name}...")
    engine = create_engine(target_conn_str)

    with engine.begin() as conn:
        df.to_sql(
            name=table_name,
            con=conn,
            if_exists=if_exists,
            index=False,
            dtype={
                'order_id': String(50),
                'customer_name': String(100),
                'product': String(200),
                'amount': Float(),
                'status': String(20),
                'order_date': DateTime(),
                'order_date_str': String(10),
                'order_month': String(7),
                'amount_level': String(10),
            }
        )

    logger.info(f"[Load] 加载完成，共写入 {len(df)} 行到 {table_name}")


# ========== 主流程 ==========

def run_etl():
    """
    执行完整的 ETL 流程
    """
    # 数据库连接信息（这里用 SQLite 做演示，换 MySQL 只需改连接串）
    SOURCE_DB = "sqlite:///source_orders.db"
    TARGET_DB = "sqlite:///dw_orders.db"

    # 源数据查询 SQL
    EXTRACT_SQL = """
        SELECT order_id, customer_name, product, amount, status, order_date
        FROM orders
        WHERE order_date >= date('now', '-30 days')
    """

    TARGET_TABLE = "fact_orders"

    logger.info("=" * 50)
    logger.info("ETL 流程开始")
    logger.info("=" * 50)

    try:
        # Step 1: Extract
        raw_data = extract(SOURCE_DB, EXTRACT_SQL)

        if raw_data.empty:
            logger.warning("源数据为空，跳过本次 ETL")
            return

        # Step 2: Transform
        clean_data = transform(raw_data)

        # Step 3: Load
        load(TARGET_DB, TARGET_TABLE, clean_data, if_exists='append')

        logger.info("=" * 50)
        logger.info("ETL 流程成功完成 ✅")
        logger.info("=" * 50)

    except Exception as e:
        logger.error(f"ETL 流程失败: {e}")
        raise


# ========== 辅助：初始化测试数据 ==========

def init_test_data():
    """在 source 库中生成测试数据供演示"""
    engine = create_engine("sqlite:///source_orders.db")

    # 生成 100 条示例订单
    np = __import__('numpy', globals(), locals(), [], 0)
    records = []
    for i in range(1, 101):
        records.append({
            'order_id': f'ORD{i:05d}',
            'customer_name': f'客户{i}' if i % 10 != 0 else None,
            'product': ['手机', '电脑', '耳机', '键盘', '鼠标'][i % 5],
            'amount': round(abs(np.random.normal(300, 150)), 2),
            'status': np.random.choice(['pending', 'completed', 'cancelled', '待处理'], p=[0.3, 0.5, 0.1, 0.1]),
            'order_date': datetime.now() - timedelta(days=np.random.randint(0, 60))
        })

    df_test = pd.DataFrame(records)
    df_test.to_sql('orders', engine, if_exists='replace', index=False)
    print("测试数据已写入 source_orders.db")


if __name__ == '__main__':
    # 首次运行先初始化测试数据
    import os
    if not os.path.exists('source_orders.db'):
        init_test_data()

    run_etl()

python etl_demo.py

2026-06-22 10:00:00 - INFO - ==================================================
2026-06-22 10:00:00 - INFO - ETL 流程开始
2026-06-22 10:00:00 - INFO - ==================================================
2026-06-22 10:00:00 - INFO - [Extract] 开始从源库抽取数据...
2026-06-22 10:00:00 - INFO - [Extract] 抽取完成，共 100 行
2026-06-22 10:00:00 - INFO - [Transform] 开始清洗数据，原始行数: 100
2026-06-22 10:00:00 - INFO -   去重: 100 → 98（删除 2 行）
2026-06-22 10:00:00 - INFO - [Transform] 清洗完成，最终行数: 98
2026-06-22 10:00:00 - INFO - [Load] 开始加载数据到目标表 fact_orders...
2026-06-22 10:00:00 - INFO - [Load] 加载完成，共写入 98 行到 fact_orders
2026-06-22 10:00:00 - INFO - ==================================================
2026-06-22 10:00:00 - INFO - ETL 流程成功完成 ✅
2026-06-22 10:00:00 - INFO - ==================================================

# 建议记录的关键指标
etl_metrics = {
    "source_rows": 10000,       # 源数据行数
    "dropped_rows": 150,        # 丢弃行数
    "loaded_rows": 9850,        # 加载行数
    "duration_sec": 45.2,       # 耗时
    "status": "success",        # 状态
}

# 推荐用一张独立的 watermark 表管理增量点位
def get_watermark(table_name: str) -> datetime:
    """获取上次同步的最大时间戳"""
    query = f"SELECT max_value FROM etl_watermark WHERE table_name = '{table_name}'"
    ...

def set_watermark(table_name: str, max_value: datetime):
    """更新水印"""
    ...

❌ 不要在 Transform 阶段做太重的计算——ETL 引擎不是数仓
❌ 不要忽略源表 schema 变更——上游加了一列，你的 ETL 可能就崩了
❌ 不要把密码写在代码里——用环境变量或密钥管理服务
❌ 不要一次性加载所有数据——分批处理，控制内存
✅ 核心表一定要做数据质量监控
✅ 每个 ETL 任务都要有回溯重跑能力
✅ 尽量保持 ETL 逻辑简单——"能跑"不等于"好维护"