4.3 数据处理与分析循环：批量清洗、探索分析与管道监控的自动化

导读：数据驱动是现代软件的核心特征，但数据处理环节往往充满重复和手工操作。本节展示如何将批量清洗、探索性分析和管道监控转化为Agent自主循环，实现数据质量问题的自动发现与修复。

学习目标

• 理解数据处理循环的三种场景：批量清洗、探索分析、管道监控
• 掌握规则驱动与AI增强结合的数据清洗循环框架
• 构建探索性数据分析的循环决策机制
• 实现数据管道异常检测与自动修复策略
• 设计数据质量评估的多维度指标体系

核心概念

数据处理循环将传统ETL中的手工步骤转变为Agent自主循环，遵循discover→plan→execute→verify四阶段：

三种循环对比：

环境准备

前置知识：Python数据处理（pandas）、SQL基础、数据质量概念

工具安装：

pip install pandas numpy
pip install openpyxl  # Excel文件支持
pip install requests  # API调用
​

分步实战

步骤一：批量数据清洗循环

"""data_cleaning_loop.py — 数据清洗循环引擎"""
import pandas as pd
import numpy as np
import json, time, re
from dataclasses import dataclass, field
from enum import Enum
from pathlib import Path
from typing import Callable


class QualityDimension(Enum):
    COMPLETENESS = "completeness"  # 完整性
    CONSISTENCY = "consistency"    # 一致性
    ACCURACY = "accuracy"          # 准确性
    UNIQUENESS = "uniqueness"      # 唯一性
    TIMELINESS = "timeliness"      # 时效性


@dataclass
class QualityRule:
    """数据质量规则"""
    name: str; dimension: QualityDimension
    check_fn: Callable  # 返回 (passed, issue_count, details)
    fix_fn: Callable = None  # 可选的修复函数


@dataclass
class QualityReport:
    score: float; dimensions: dict[str, float]
    issues: list[dict]; total_rows: int; cleaned_rows: int


class DataCleaningLoop:
    """
    数据清洗循环：
    发现问题 → 匹配规则 → 执行清洗 → 验证 → 循环或完成
    """
    
    def __init__(self, max_loops: int = 5, target_score: float = 95.0):
        self.max_loops = max_loops
        self.target_score = target_score
        self.rules: list[QualityRule] = []
        self._register_default_rules()
    
    def _register_default_rules(self):
        """注册默认数据质量规则"""
        
        # 完整性：空值检测
        self.rules.append(QualityRule(
            "null_check", QualityDimension.COMPLETENESS,
            check_fn=lambda df: self._check_nulls(df),
            fix_fn=lambda df: self._fill_nulls(df)
        ))
        
        # 一致性：格式验证（邮箱）
        self.rules.append(QualityRule(
            "email_format", QualityDimension.CONSISTENCY,
            check_fn=lambda df: self._check_email_format(df),
            fix_fn=lambda df: self._fix_email_format(df)
        ))
        
        # 唯一性：重复行检测
        self.rules.append(QualityRule(
            "duplicate_check", QualityDimension.UNIQUENESS,
            check_fn=lambda df: self._check_duplicates(df),
            fix_fn=lambda df: df.drop_duplicates(keep='first')
        ))
        
        # 准确性：范围检测
        self.rules.append(QualityRule(
            "range_check", QualityDimension.ACCURACY,
            check_fn=lambda df: self._check_ranges(df)
        ))
        
        # 一致性：类型标准化
        self.rules.append(QualityRule(
            "type_standard", QualityDimension.CONSISTENCY,
            check_fn=lambda df: self._check_types(df),
            fix_fn=lambda df: self._standardize_types(df)
        ))
    
    def run(self, df: pd.DataFrame) -> tuple[pd.DataFrame, QualityReport]:
        """执行数据清洗循环"""
        print(f"🧹 数据清洗循环 | {len(df)}行 × {len(df.columns)}列")
        original_count = len(df)
        
        for loop_num in range(1, self.max_loops + 1):
            print(f"\n{'='*50}")
            print(f"🔄 清洗循环 #{loop_num}")
            
            # DISCOVER + PLAN: 检查所有质量规则
            issues = []
            for rule in self.rules:
                passed, count, details = rule.check_fn(df)
                if not passed:
                    issues.append({
                        "rule": rule.name,
                        "dimension": rule.dimension.value,
                        "count": count, "details": details
                    })
                    print(f"  ❌ {rule.name}: {count}个问题")
            
            if not issues:
                score = 100.0
                print("  ✅ 所有规则通过！")
                break
            
            # 计算当前质量评分
            score = self._calculate_score(df, self.rules)
            print(f"  📊 质量评分: {score:.1f}%")
            
            if score >= self.target_score:
                print(f"  ✅ 质量达标（{score:.1f}% ≥ {self.target_score}%）")
                break
            
            # EXECUTE: 应用修复
            for issue in issues:
                rule = next(r for r in self.rules if r.name == issue["rule"])
                if rule.fix_fn:
                    print(f"  🔧 修复 {rule.name}...")
                    df = rule.fix_fn(df)
            
            # VERIFY: 检查评分变化
            new_score = self._calculate_score(df, self.rules)
            improvement = new_score - score
            print(f"  📈 评分变化: {score:.1f}% → {new_score:.1f}% (+{improvement:.1f}%)")
            
            if improvement < 0.5:
                print(f"  ⚠️ 改善停滞，退出循环")
                score = new_score
                break
        
        report = self._generate_report(df, self.rules, original_count)
        return df, report
    
    # ——— 规则实现 ———
    
    def _check_nulls(self, df):
        null_counts = df.isnull().sum()
        issues = null_counts[null_counts > 0]
        return (len(issues) == 0, int(issues.sum()), issues.to_dict())
    
    def _fill_nulls(self, df):
        for col in df.columns:
            if df[col].isnull().any():
                if df[col].dtype in ['int64', 'float64']:
                    df[col] = df[col].fillna(df[col].median())
                else:
                    df[col] = df[col].fillna("UNKNOWN")
        return df
    
    def _check_email_format(self, df):
        email_cols = [c for c in df.columns if 'email' in c.lower()]
        issues = {}
        for col in email_cols:
            if col in df.columns:
                bad = df[col].apply(lambda x: not bool(re.match(
                    r'^[\w.-]+@[\w.-]+\.\w+$', str(x))) if pd.notna(x) else False)
                count = int(bad.sum())
                if count > 0:
                    issues[col] = count
        return (len(issues) == 0, sum(issues.values()), issues)
    
    def _fix_email_format(self, df):
        for col in df.columns:
            if 'email' in col.lower():
                df[col] = df[col].apply(
                    lambda x: str(x).strip().lower() if pd.notna(x) else x)
        return df
    
    def _check_duplicates(self, df):
        dup = df.duplicated().sum()
        return (dup == 0, int(dup), {"duplicate_rows": int(dup)})
    
    def _check_ranges(self, df):
        issues = {}
        numeric_cols = df.select_dtypes(include=[np.number]).columns
        for col in numeric_cols:
            q1, q3 = df[col].quantile([0.25, 0.75])
            iqr = q3 - q1
            outliers = ((df[col] < q1 - 3*iqr) | (df[col] > q3 + 3*iqr)).sum()
            if outliers > 0:
                issues[col] = {"outliers": int(outliers), "range": [float(q1-3*iqr), float(q3+3*iqr)]}
        return (len(issues) == 0, sum(v["outliers"] for v in issues.values()), issues)
    
    def _check_types(self, df):
        issues = {}
        for col in df.columns:
            if df[col].dtype == object:
                mixed = df[col].apply(type).nunique() > 2
                if mixed:
                    issues[col] = "mixed_types"
        return (len(issues) == 0, len(issues), issues)
    
    def _standardize_types(self, df):
        for col in df.columns:
            if df[col].dtype == object:
                try: df[col] = pd.to_numeric(df[col])
                except: pass
        return df
    
    def _calculate_score(self, df, rules):
        total = len(rules)
        passed = 0
        for rule in rules:
            ok, _, _ = rule.check_fn(df)
            if ok: passed += 1
        return (passed / total) * 100 if total > 0 else 0.0
    
    def _generate_report(self, df, rules, original_count):
        score = self._calculate_score(df, rules)
        return QualityReport(
            score=score,
            dimensions={r.dimension.value: 100.0 if r.check_fn(df)[0] else 50.0 for r in rules},
            issues=[], total_rows=original_count, cleaned_rows=len(df)
        )


# 使用示例
if __name__ == "__main__":
    # 模拟脏数据
    data = {
        "name": ["Alice", "Bob", None, "Alice", "Eve"],
        "email": ["alice@test.com", "BOB@TEST.COM", "invalid", "alice@test.com", "eve@test"],
        "age": [25, 30, 28, 25, 999],
        "salary": [50000, None, 45000, 50000, 38000]
    }
    df = pd.DataFrame(data)
    
    cleaner = DataCleaningLoop(target_score=90.0)
    clean_df, report = cleaner.run(df)
    print(f"\n清洗完成: {report.score:.1f}% | {report.total_rows}→{report.cleaned_rows}行")
​

步骤二：探索性数据分析循环

"""eda_loop.py — 探索性数据分析循环"""
import pandas as pd
import numpy as np
from dataclasses import dataclass
from enum import Enum


class InsightType(Enum):
    DISTRIBUTION = "distribution"     # 分布异常
    CORRELATION = "correlation"       # 强相关
    OUTLIER = "outlier"               # 离群值
    TREND = "trend"                   # 趋势变化
    SEGMENT = "segment"              # 分群差异


@dataclass
class Insight:
    type: InsightType; description: str
    columns: list[str]; value: float = 0.0


class EDALoop:
    """
    探索性分析循环：
    定义目标 → 分析 → 评估价值 → 深入或调整方向
    """
    
    def __init__(self, max_rounds: int = 6):
        self.max_rounds = max_rounds
        self.insights: list[Insight] = []
    
    def explore(self, df: pd.DataFrame, objective: str = "auto") -> list[Insight]:
        """执行探索性分析循环"""
        print(f"📊 EDA循环 | 目标: {objective}")
        
        for rnd in range(1, self.max_rounds + 1):
            print(f"\n🔄 分析轮次 #{rnd}")
            new_insights = []
            
            # 阶段1: 基础统计
            new_insights.extend(self._check_distributions(df))
            new_insights.extend(self._check_correlations(df))
            new_insights.extend(self._check_outliers(df))
            
            # 阶段2: 深入分析（根据已有发现）
            if self.insights:
                for prev in self.insights[-3:]:
                    new_insights.extend(self._deep_dive(df, prev))
            
            # 阶段3: 评估价值
            valuable = [i for i in new_insights if i.value >= 0.5]
            print(f"  发现 {len(new_insights)} 个洞察，{len(valuable)} 个有价值的")
            
            for ins in valuable:
                print(f"  💡 [{ins.type.value}] {ins.description} (强度:{ins.value:.2f})")
                self.insights.append(ins)
            
            if len(valuable) == 0:
                print(f"  📋 分析饱和，退出循环")
                break
        
        return self.insights
    
    def _check_distributions(self, df) -> list[Insight]:
        """检查数据分布特征"""
        insights = []
        numeric = df.select_dtypes(include=[np.number]).columns
        for col in numeric[:10]:
            skew = df[col].skew()
            if abs(skew) > 1.5:
                insights.append(Insight(
                    InsightType.DISTRIBUTION,
                    f"{col} 高度偏斜 (skew={skew:.2f})",
                    [col], min(abs(skew)/3, 1.0)))
        return insights
    
    def _check_correlations(self, df) -> list[Insight]:
        """检查特征相关性"""
        insights = []
        numeric = df.select_dtypes(include=[np.number])
        if len(numeric.columns) < 2: return insights
        corr = numeric.corr()
        for i in range(len(corr.columns)):
            for j in range(i+1, len(corr.columns)):
                r = abs(corr.iloc[i, j])
                if r > 0.7:
                    insights.append(Insight(
                        InsightType.CORRELATION,
                        f"{corr.columns[i]} 与 {corr.columns[j]} 强相关 (r={r:.2f})",
                        [corr.columns[i], corr.columns[j]], r))
        return insights
    
    def _check_outliers(self, df) -> list[Insight]:
        """检测离群值"""
        insights = []
        numeric = df.select_dtypes(include=[np.number]).columns
        for col in numeric[:10]:
            q1, q3 = df[col].quantile([0.25, 0.75])
            iqr = q3 - q1
            outlier_pct = ((df[col] < q1-3*iqr) | (df[col] > q3+3*iqr)).mean()
            if outlier_pct > 0.02:
                insights.append(Insight(
                    InsightType.OUTLIER,
                    f"{col} 有 {outlier_pct:.1%} 离群值",
                    [col], min(outlier_pct*10, 1.0)))
        return insights
    
    def _deep_dive(self, df, insight: Insight) -> list[Insight]:
        """基于已有发现深入分析"""
        if insight.type == InsightType.CORRELATION and len(insight.columns) >= 2:
            c1, c2 = insight.columns[0], insight.columns[1]
            if c2 in df.columns and df[c2].dtype == object:
                groups = df.groupby(c2)[c1].mean()
                if groups.std() / groups.mean() > 0.3:
                    return [Insight(InsightType.SEGMENT,
                        f"{c1} 在 {c2} 不同分组中差异显著",
                        insight.columns, 0.6)]
        return []


if __name__ == "__main__":
    np.random.seed(42)
    data = {
        "revenue": np.random.lognormal(10, 1, 1000),
        "users": np.random.normal(1000, 200, 1000),
        "region": np.random.choice(["A","B","C"], 1000),
        "satisfaction": np.random.uniform(1, 5, 1000)
    }
    df = pd.DataFrame(data)
    
    eda = EDALoop(max_rounds=4)
    insights = eda.explore(df)
    print(f"\n总计 {len(insights)} 个有价值的洞察")
​

步骤三：数据管道监控循环

"""pipeline_monitor.py — 数据管道监控与自动修复"""
import time, json
from dataclasses import dataclass, field
from enum import Enum
from typing import Callable


class AlertLevel(Enum):
    INFO = "info"; WARNING = "warning"; CRITICAL = "critical"


@dataclass
class Anomaly:
    metric: str; expected: float; actual: float
    deviation_pct: float; level: AlertLevel
    timestamp: float = field(default_factory=time.time)


@dataclass
class FixResult:
    success: bool; action: str; details: str


class PipelineMonitor:
    """
    数据管道监控循环：
    检测异常 → 分析根因 → 尝试修复 → 告警/升级
    """
    
    def __init__(self, alert_threshold: float = 20.0):
        self.alert_threshold = alert_threshold  # 偏差百分比阈值
        self.anomalies: list[Anomaly] = []
        self.baseline: dict[str, float] = {}  # 基线指标
    
    def set_baseline(self, metrics: dict[str, float]):
        self.baseline = metrics
    
    def check(self, current_metrics: dict[str, float]) -> list[Anomaly]:
        """检测异常"""
        anomalies = []
        for metric, value in current_metrics.items():
            if metric in self.baseline:
                expected = self.baseline[metric]
                if expected != 0:
                    deviation = abs(value - expected) / expected * 100
                    if deviation >= self.alert_threshold:
                        level = (AlertLevel.CRITICAL if deviation >= 50
                                else AlertLevel.WARNING)
                        anomaly = Anomaly(
                            metric, expected, value, deviation, level)
                        anomalies.append(anomaly)
                        print(f"  {'🚨' if level==AlertLevel.CRITICAL else '⚠️'} "
                              f"{metric}: {expected}→{value} ({deviation:+.1f}%)")
                        self.anomalies.append(anomaly)
        return anomalies
    
    def auto_fix(self, anomaly: Anomaly, fix_fn: Callable) -> FixResult:
        """尝试自动修复"""
        print(f"  🔧 尝试修复 {anomaly.metric}...")
        try:
            result = fix_fn(anomaly)
            if result.get("success"):
                self.baseline[anomaly.metric] = result.get("new_value", anomaly.actual)
                return FixResult(True, result.get("action",""), str(result))
        except Exception as e:
            return FixResult(False, "auto_fix_error", str(e))
        return FixResult(False, "fix_failed", "修复函数返回失败")


if __name__ == "__main__":
    monitor = PipelineMonitor(alert_threshold=15.0)
    monitor.set_baseline({
        "row_count": 10000, "null_pct": 2.0,
        "processing_time": 30.0, "error_rate": 0.5
    })
    
    current = {"row_count": 7500, "null_pct": 8.5, 
                "processing_time": 120.0, "error_rate": 0.3}
    
    anomalies = monitor.check(current)
    print(f"\n检测到 {len(anomalies)} 个异常")
    
    for a in anomalies:
        fix = monitor.auto_fix(a, lambda a: {
            "success": True, "action": "调整参数",
            "new_value": a.actual
        })
        print(f"  修复结果: {fix.success}")
​

完整示例

"""data_pipeline_loop.py — 完整数据处理管道循环"""
from data_cleaning_loop import DataCleaningLoop
from eda_loop import EDALoop
from pipeline_monitor import PipelineMonitor
import pandas as pd

class DataPipelineLoop:
    """整合清洗→分析→监控的完整管道"""
    
    def __init__(self, monitor_threshold=15.0):
        self.cleaner = DataCleaningLoop(target_score=90.0)
        self.eda = EDALoop(max_rounds=4)
        self.monitor = PipelineMonitor(alert_threshold=monitor_threshold)
    
    def process(self, raw_data, objective="auto"):
        # 阶段1: 清洗
        print("\n" + "="*60)
        print("阶段1: 数据清洗")
        clean_data, report = self.cleaner.run(raw_data)
        
        # 阶段2: 分析
        print("\n" + "="*60)
        print("阶段2: 探索性分析")
        insights = self.eda.explore(clean_data, objective)
        
        # 阶段3: 监控基线
        self._update_baseline(clean_data)
        
        return {"clean_data": clean_data, "quality": report.score,
                "insights": len(insights)}

if __name__ == "__main__":
    pipeline = DataPipelineLoop()
    df = pd.DataFrame({"col1": [1,2,None,1,5], "col2": ["a","B",None,"a","c"]})
    result = pipeline.process(df)
    print(f"\n结果: 质量={result['quality']:.1f}%, 洞察={result['insights']}个")
​

常见问题FAQ

Q1: 数据清洗循环会不会过度清洗，丢失有效信息？
A: 确实有风险。保障措施：清洗前备份数据、记录每步清洗操作的完整日志、对关键列设置"不可自动清洗"白名单、清洗后进行抽样人工审核。特别是填充空值时，median/mean填充可能改变分布特征。

Q2: EDA循环如何避免陷入无意义的分析？
A: 设置价值评估阈值（insight.value >= 0.5才保留）、限制分析轮次（max_rounds=4-6）、设置冷却期（连续2轮无新发现则退出）、记录已分析的维度避免重复。核心是让Agent评估"这个发现是否值得深入"。

Q3: 管道监控如何区分正常波动和真正的异常？
A: 使用动态基线而非固定阈值：基于最近N个时间窗口的统计分布，只有当前值超出3σ才判定为异常。季节性数据需要考虑时间模式。建议结合统计方法和业务规则双重判定。

最佳实践与避坑

✅ 备份原始数据 — 任何清洗操作前必须备份
✅ 清洗日志完整 — 记录每步的操作和影响范围
✅ 质量评分量化 — 用多维度评分而非单一指标
✅ EDA有预算 — 限制分析轮次和计算资源消耗
✅ 监控基线动态 — 用滑动窗口而非固定基线

⚠️ 不要自动删除数据 — 标记而非删除可疑数据
⚠️ 不要忽略数据语义 — 纯统计规则可能不符合业务逻辑
⚠️ EDA不要无限探索 — 设置轮次上限和冷却机制

本节小结

数据处理循环展示了Loop Engineering在非代码场景的强大应用。批量清洗循环通过规则引擎实现质量问题自动发现和修复，EDA循环让Agent自主探索数据并评估发现价值，管道监控循环实现异常的实时检测与修复建议。

关键要点：数据清洗需要备份和日志；EDA需要价值评估防止无限探索；监控需要动态基线适应数据变化。

下一节过渡

数据处理循环处理结构化数据，而内容生产是另一种常见场景。下一节探讨内容生产与发布循环，了解如何让Agent自动化文档生成、质量检查和SEO优化流程。

延伸阅读

1. pandas官方文档 — 数据清洗API完整参考
2. Great Expectations — 数据质量验证框架
3. 《Python数据科学手册》— EDA方法论
4. dbt文档 — 数据管道工程化实践

关键词： 数据处理循环、数据清洗自动化、EDA探索分析、数据管道监控、数据质量评分、Agent驱动数据工程、Loop Engineering数据场景
难度： ⭐⭐⭐ 中级
预计阅读时间： 18分钟