ETL进阶：数据清洗与转换实战

# 缺失值的多种表现
missing_patterns = {
    "email": [None, "", "N/A", "NULL", "unknown@unknown.com"],
    "age": [None, 0, -1, 999],
    "date": [None, "", "1900-01-01", "1970-01-01"]
}

import pandas as pd
import numpy as np

def detect_outliers_iqr(df: pd.DataFrame, column: str) -> pd.Series:
    """基于 IQR 方法检测异常值"""
    Q1 = df[column].quantile(0.25)
    Q3 = df[column].quantile(0.75)
    IQR = Q3 - Q1
    lower = Q1 - 1.5 * IQR
    upper = Q3 + 1.5 * IQR
    return (df[column] < lower) | (df[column] > upper)

def detect_outliers_business(df: pd.DataFrame, column: str, 
                              min_val: float, max_val: float) -> pd.Series:
    """基于业务规则的异常值检测"""
    return (df[column] < min_val) | (df[column] > max_val)

# 多层去重策略
def deduplicate(df: pd.DataFrame, keys: list[str]) -> pd.DataFrame:
    # 第一步：完全重复去除
    df = df.drop_duplicates()
    # 第二步：基于业务主键去重，保留最新记录
    df = df.sort_values("updated_at").drop_duplicates(subset=keys, keep="last")
    return df

date_formats = {
    "source_a": "%Y-%m-%d",
    "source_b": "%m/%d/%Y",
    "source_c": "%Y年%m月%d日"
}

def normalize_date(value: str, source: str) -> str:
    fmt = date_formats.get(source)
    if fmt:
        dt = datetime.strptime(value.strip(), fmt)
        return dt.strftime("%Y-%m-%d")  # 统一输出
    raise ValueError(f"Unknown date format for source: {source}")

from abc import ABC, abstractmethod
from dataclasses import dataclass, field
from typing import Optional, Any
import logging

logger = logging.getLogger(__name__)


@dataclass
class ValidationResult:
    row_id: Optional[str]
    field: str
    severity: str       # fatal / error / warning / info
    message: str
    raw_value: Any


class BaseCleaner(ABC):
    """清洗器基类——每个清洗规则是一个独立单元"""

    def __init__(self, name: str, severity: str = "error"):
        self.name = name
        self.severity = severity

    @abstractmethod
    def validate(self, df: pd.DataFrame) -> list[ValidationResult]:
        ...

    @abstractmethod
    def fix(self, df: pd.DataFrame) -> pd.DataFrame:
        ...

    def run(self, df: pd.DataFrame) -> tuple[pd.DataFrame, list[ValidationResult]]:
        """执行校验 + 修复"""
        issues = self.validate(df)
        df_clean = self.fix(df)
        for issue in issues:
            log_func = logger.warning if issue.severity in ("error",) else logger.info
            log_func(f"[{issue.severity.upper()}] {issue.field}: {issue.message}")
        return df_clean, issues

class CleaningPipeline:
    def __init__(self):
        self._cleaners: list[BaseCleaner] = []
        self._rejects: list[ValidationResult] = []

    def add_cleaner(self, cleaner: BaseCleaner):
        self._cleaners.append(cleaner)

    def run(self, df: pd.DataFrame, reject_on_fatal: bool = True) -> pd.DataFrame:
        all_issues = []
        for cleaner in self._cleaners:
            df, issues = cleaner.run(df)
            all_issues.extend(issues)

        # 致命错误记录拒绝
        if reject_on_fatal:
            fatal_issues = [i for i in all_issues if i.severity == "fatal"]
            if fatal_issues:
                self._rejects.extend(fatal_issues)
                fatal_ids = {i.row_id for i in fatal_issues if i.row_id}
                df = df[~df.index.isin(fatal_ids)]
                logger.warning(f"Rejected {len(fatal_ids)} records due to fatal errors")

        return df

def safe_type_cast(df: pd.DataFrame, mapping: dict[str, str]) -> pd.DataFrame:
    """安全的类型转换，失败时置空并告警"""
    for col, target_type in mapping.items():
        if col not in df.columns:
            continue
        try:
            if target_type == "int":
                df[col] = pd.to_numeric(df[col], errors="coerce").astype("Int64")
            elif target_type == "float":
                df[col] = pd.to_numeric(df[col], errors="coerce")
            elif target_type == "datetime":
                df[col] = pd.to_datetime(df[col], errors="coerce")
            elif target_type == "str":
                df[col] = df[col].astype(str)
        except Exception as e:
            logger.error(f"Type cast failed for {col}: {e}")
    return df

# 使用示例
type_mapping = {
    "order_id": "int",
    "amount": "float",
    "created_at": "datetime",
    "user_name": "str"
}
df = safe_type_cast(df, type_mapping)

FIELD_MAPPING = {
    "src_order_id": "order_id",
    "src_user_id": "user_id",
    "order_amount": "amount",
    "order_date": "created_at",
}

def apply_field_mapping(df: pd.DataFrame, mapping: dict[str, str]) -> pd.DataFrame:
    return df.rename(columns=mapping)

VALUE_MAPPING = {
    "gender": {
        "M": "male", "F": "female",
        "1": "male", "0": "female",
        "男": "male", "女": "female"
    },
    "status": {
        "1": "active", "2": "inactive", "3": "deleted",
        "Y": "active", "N": "inactive"
    }
}

def apply_value_mapping(df: pd.DataFrame, mapping: dict) -> pd.DataFrame:
    for col, mapper in mapping.items():
        if col in df.columns:
            df[col] = df[col].map(mapper).fillna(df[col])
    return df

def add_derived_fields(df: pd.DataFrame) -> pd.DataFrame:
    """添加业务常用的衍生字段"""
    # 订单金额分类
    df["amount_category"] = pd.cut(
        df["amount"],
        bins=[0, 100, 1000, 10000, float("inf")],
        labels=["小额", "中额", "大额", "超大额"]
    )
    # 时间维度拆解
    if "created_at" in df.columns:
        df["order_year"] = df["created_at"].dt.year
        df["order_month"] = df["created_at"].dt.month
        df["order_weekday"] = df["created_at"].dt.weekday
        df["is_weekend"] = df["order_weekday"].isin([5, 6])
    # 用户首单标记（假设数据已按 user_id 排序）
    df["is_first_order"] = ~df["user_id"].duplicated()
    return df

def incremental_extract_timestamp(
    conn, table: str, last_run: datetime, batch_size: int = 10000
) -> pd.DataFrame:
    """基于时间戳的增量抽取"""
    query = f"""
        SELECT * FROM {table}
        WHERE updated_at >= %(last_run)s
        ORDER BY updated_at ASC
        LIMIT %(batch_size)s
    """
    df = pd.read_sql(query, conn, params={
        "last_run": last_run,
        "batch_size": batch_size
    })
    return df

# 以 Debezium + Kafka 为例的伪代码示意
def consume_cdc_events(topic: str, bootstrap_servers: str):
    """消费 CDC 事件流"""
    consumer = KafkaConsumer(
        topic,
        bootstrap_servers=bootstrap_servers,
        value_deserializer=lambda m: json.loads(m.decode("utf-8")),
        auto_offset_reset="earliest",
        enable_auto_commit=False,
    )
    for message in consumer:
        event = message.value
        if event["op"] == "c":      # Create
            upsert_to_target(event["after"])
        elif event["op"] == "u":    # Update
            upsert_to_target(event["after"])
        elif event["op"] == "d":    # Delete
            soft_delete_in_target(event["before"]["id"])
        consumer.commit()

def incremental_from_api_logs(log_file: str, parser: callable) -> Generator:
    """解析业务操作日志，还原数据变更"""
    with open(log_file, "r") as f:
        for line in tail(f):  # tail -f 类似效果
            event = parser(line)
            if event["action"] in ("create", "update"):
                yield fetch_full_record(event["record_id"])

#!/usr/bin/env python3
"""
etl_pipeline.py —— 通用 ETL 数据清洗管道
支持：多层校验、异常隔离、日志链路追踪、断点续传
"""

import hashlib
import json
import logging
import sys
import time
from datetime import datetime, timedelta
from pathlib import Path
from typing import Optional

import pandas as pd
import yaml

# ── 日志配置 ────────────────────────────────────────────
def setup_logger(name: str, log_dir: str = "logs") -> logging.Logger:
    Path(log_dir).mkdir(parents=True, exist_ok=True)
    logger = logging.getLogger(name)
    logger.setLevel(logging.DEBUG)

    fmt = logging.Formatter(
        "[%(asctime)s] %(levelname)-8s [%(name)s] %(message)s",
        datefmt="%Y-%m-%d %H:%M:%S"
    )
    # 文件日志（全量）
    fh = logging.FileHandler(f"{log_dir}/{name}_{datetime.now():%Y%m%d}.log")
    fh.setLevel(logging.DEBUG)
    fh.setFormatter(fmt)
    # 控制台日志（INFO 以上）
    ch = logging.StreamHandler()
    ch.setLevel(logging.INFO)
    ch.setFormatter(fmt)

    logger.addHandler(fh)
    logger.addHandler(ch)
    return logger


logger = setup_logger("etl_pipeline")


# ── 数据校验器 ──────────────────────────────────────────
class DataValidator:
    """数据校验规则集合"""

    @staticmethod
    def check_not_null(df: pd.DataFrame, columns: list[str]) -> pd.DataFrame:
        """必填字段校验——空值所在行标记为 reject"""
        mask = df[columns].isna().any(axis=1)
        df.loc[mask, "_reject_reason"] = df.loc[mask, "_reject_reason"].fillna("") + \
            f"missing_required:{','.join(columns)};"
        return df

    @staticmethod
    def check_unique(df: pd.DataFrame, keys: list[str]) -> pd.DataFrame:
        """唯一性校验"""
        dup_mask = df.duplicated(subset=keys, keep=False)
        df.loc[dup_mask, "_reject_reason"] = df.loc[dup_mask, "_reject_reason"].fillna("") + \
            f"duplicate_key:{','.join(keys)};"
        return df

    @staticmethod
    def check_value_range(
        df: pd.DataFrame, column: str, min_val: float = None, max_val: float = None
    ) -> pd.DataFrame:
        """值域校验"""
        if min_val is not None:
            mask = df[column] < min_val
            df.loc[mask, "_reject_reason"] = df.loc[mask, "_reject_reason"].fillna("") + \
                f"below_min:{column}={min_val};"
        if max_val is not None:
            mask = df[column] > max_val
            df.loc[mask, "_reject_reason"] = df.loc[mask, "_reject_reason"].fillna("") + \
                f"above_max:{column}={max_val};"
        return df


# ── 清洗管道主类 ────────────────────────────────────────
class ETLPipeline:
    """
    可配置的 ETL 数据清洗管道

    用法：
        pipeline = ETLPipeline(config_path="pipeline_config.yaml")
        pipeline.run("orders_20240617.csv")
    """

    def __init__(self, config_path: str):
        with open(config_path, "r") as f:
            self.config = yaml.safe_load(f)
        self.logger = logging.getLogger(f"etl_pipeline.{self.config.get('name', 'default')}")
        self._stats = {"total": 0, "passed": 0, "rejected": 0, "duration": 0.0}

    def _load_data(self, path: str, **kwargs) -> pd.DataFrame:
        """多格式数据加载"""
        ext = Path(path).suffix.lower()
        loaders = {
            ".csv": pd.read_csv,
            ".json": pd.read_json,
            ".parquet": pd.read_parquet,
            ".xlsx": pd.read_excel,
        }
        loader = loaders.get(ext)
        if not loader:
            raise ValueError(f"Unsupported file format: {ext}")
        self.logger.info(f"Loading data from {path}")
        df = loader(path, **kwargs)
        df["_reject_reason"] = ""
        self._stats["total"] = len(df)
        self.logger.info(f"Loaded {len(df)} records")
        return df

    def _validate(self, df: pd.DataFrame) -> pd.DataFrame:
        """执行校验规则"""
        rules = self.config.get("validation", {})
        validator = DataValidator()

        for rule in rules.get("not_null", []):
            df = validator.check_not_null(df, rule["columns"])
            self.logger.debug(f"Not-null check: {rule['columns']}")

        for rule in rules.get("unique", []):
            df = validator.check_unique(df, rule["keys"])
            self.logger.debug(f"Unique check: {rule['keys']}")

        for rule in rules.get("range", []):
            df = validator.check_value_range(
                df, rule["column"], rule.get("min"), rule.get("max")
            )
            self.logger.debug(f"Range check: {rule['column']}")

        return df

    def _transform(self, df: pd.DataFrame) -> pd.DataFrame:
        """执行转换逻辑"""
        transforms = self.config.get("transform", {})

        # 类型转换
        for col, dtype in transforms.get("type_cast", {}).items():
            if col in df.columns:
                try:
                    df[col] = df[col].astype(dtype)
                except Exception as e:
                    self.logger.warning(f"Type cast failed for {col}: {e}")

        # 字段映射
        rename_map = transforms.get("field_mapping", {})
        if rename_map:
            df = df.rename(columns=rename_map)

        # 值映射
        for col, mapping in transforms.get("value_mapping", {}).items():
            if col in df.columns:
                df[col] = df[col].map(mapping).fillna(df[col])

        self.logger.info("Transform completed")
        return df

    def _split_rejects(self, df: pd.DataFrame) -> tuple[pd.DataFrame, pd.DataFrame]:
        """将有拒绝原因的记录分离出来"""
        reject_mask = df["_reject_reason"].str.len() > 0
        rejects = df[reject_mask].copy()
        passed = df[~reject_mask].copy()
        passed = passed.drop(columns=["_reject_reason"])
        rejects = rejects.drop(columns=["_reject_reason"])
        self._stats["rejected"] = len(rejects)
        self._stats["passed"] = len(passed)
        return passed, rejects

    def run(self, source: str, **kwargs) -> dict:
        """执行完整的 ETL 管道"""
        start_time = time.time()
        self.logger.info(f"Pipeline started — source={source}")

        try:
            # 1. 数据加载
            df = self._load_data(source, **kwargs)
            # 2. 数据校验
            df = self._validate(df)
            # 3. 数据转换
            df = self._transform(df)
            # 4. 分离合法/拒绝数据
            passed, rejects = self._split_rejects(df)
        except Exception as e:
            self.logger.exception(f"Pipeline failed: {e}")
            raise

        # 5. 输出
        output = self.config.get("output", {})
        if not passed.empty:
            passed_path = output.get("path", "output") + "/passed"
            Path(passed_path).mkdir(parents=True, exist_ok=True)
            passed.to_parquet(f"{passed_path}/{Path(source).stem}.parquet", index=False)
            self.logger.info(f"Passed records written: {len(passed)}")

        if not rejects.empty:
            reject_path = output.get("reject_path", "output") + "/rejects"
            Path(reject_path).mkdir(parents=True, exist_ok=True)
            rejects.to_csv(f"{reject_path}/{Path(source).stem}_rejects.csv", index=False)
            self.logger.warning(f"Rejected records written: {len(rejects)}")

        self._stats["duration"] = round(time.time() - start_time, 2)
        self.logger.info(
            f"Pipeline finished — "
            f"total={self._stats['total']}, "
            f"passed={self._stats['passed']}, "
            f"rejected={self._stats['rejected']}, "
            f"duration={self._stats['duration']}s"
        )
        return self._stats

# pipeline_config.yaml
name: order_etl_pipeline

validation:
  not_null:
    - columns: ["order_id", "user_id", "amount"]
  unique:
    - keys: ["order_id"]
  range:
    - column: "amount"
      min: 0
      max: 1000000

transform:
  type_cast:
    order_id: int
    amount: float
    created_at: datetime64[ns]
  field_mapping:
    order_id: order_id
    user_id: user_id
    amount: order_amount
    created_at: order_date
  value_mapping:
    status:
      "1": active
      "2": inactive

output:
  path: ./output/data
  reject_path: ./output/rejects

# 生产调用
pipeline = ETLPipeline("pipeline_config.yaml")
stats = pipeline.run("raw_orders_20240617.csv")
print(stats)
# {'total': 50000, 'passed': 49832, 'rejected': 168, 'duration': 3.42}

-- 数据质量监控表
CREATE TABLE dw_quality_metrics (
    id BIGINT AUTO_INCREMENT PRIMARY KEY,
    table_name VARCHAR(128),
    batch_id VARCHAR(64),
    run_date DATE,
    row_count INT,
    null_count INT,
    duplicate_count INT,
    reject_count INT,
    duration_seconds DECIMAL(10,2),
    status VARCHAR(20),          -- SUCCESS / WARNING / FAILED
    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);

def check_row_count_anomaly(
    table: str, current_count: int, history_days: int = 30, threshold: float = 0.2
) -> bool:
    """检测行数是否异常"""
    query = """
        SELECT AVG(row_count) as avg_rows, STDDEV(row_count) as std_rows
        FROM dw_quality_metrics
        WHERE table_name = %(table)s
          AND run_date >= NOW() - INTERVAL %(days)s DAY
          AND status = 'SUCCESS'
    """
    # 执行查询…
    avg_rows, std_rows = 485000, 12000
    lower = avg_rows * (1 - threshold)
    upper = avg_rows * (1 + threshold)
    if current_count < lower or current_count > upper:
        logger.error(f"Row count anomaly for {table}: {current_count} (expected ~{avg_rows})")
        return False
    return True

def profile_columns(df: pd.DataFrame) -> dict:
    """生成字段画像"""
    profile = {}
    for col in df.columns:
        null_pct = df[col].isna().mean()
        profile[col] = {
            "null_rate": round(null_pct, 4),
            "dtype": str(df[col].dtype),
            "unique_count": df[col].nunique(),
        }
        if null_pct > 0.05:  # 空值率超过 5% 告警
            logger.warning(f"Column {col} null rate: {null_pct:.2%}")
    return profile

# 企业微信机器人告警示例
def send_wechat_alert(webhook: str, title: str, content: str):
    payload = {
        "msgtype": "markdown",
        "markdown": {
            "content": f"## ⚠️ ETL 告警\n**{title}**\n{content}\n时间: {datetime.now()}"
        }
    }
    requests.post(webhook, json=payload, timeout=5)

# 使用 prometheus_client 暴露指标
from prometheus_client import Counter, Histogram, Gauge, start_http_server

etl_duration = Histogram(
    "etl_task_duration_seconds", "ETL task duration",
    ["task_name", "source"],
    buckets=(1, 5, 10, 30, 60, 120, 300, 600)
)
etl_status = Gauge(
    "etl_task_status", "ETL task status (0=success, 1=fail)",
    ["task_name"]
)
etl_records_rejected = Counter(
    "etl_records_rejected_total", "Total rejected records",
    ["task_name", "reason"]
)


def monitor_task(task_name: str, source: str):
    """装饰器：自动采集 ETL 监控指标"""
    def decorator(func):
        @functools.wraps(func)
        def wrapper(*args, **kwargs):
            start = time.time()
            try:
                result = func(*args, **kwargs)
                etl_status.labels(task_name=task_name).set(0)
                return result
            except Exception as e:
                etl_status.labels(task_name=task_name).set(1)
                raise
            finally:
                etl_duration.labels(
                    task_name=task_name, source=source
                ).observe(time.time() - start)
        return wrapper
    return decorator