Python脚本如何打造高效稳定数据体系

wen python案例 2026-06-24 241

本文目录导读：

数据体系架构设计
数据采集层实现
数据质量保障
数据处理优化
容错与恢复机制
监控与告警
架构最佳实践

我来详细介绍如何用Python构建高效稳定的数据体系。

数据体系架构设计

核心架构组件

# 1. 数据分层架构
from dataclasses import dataclass
from typing import Optional, Any
import logging
@dataclass
class DataLayer:
    """数据分层定义"""
    raw: str = "raw"      # 原始数据层
    staging: str = "staging"  # 清洗层
    dwd: str = "dwd"      # 明细层
    dim: str = "dim"      # 维度层
    dws: str = "dws"      # 汇总层
    app: str = "app"      # 应用层
class DataPipeline:
    """数据管道基类"""
    def __init__(self, config: dict):
        self.config = config
        self.logger = logging.getLogger(__name__)
        self.error_handler = ErrorHandler()
    def extract(self):
        """数据提取"""
        raise NotImplementedError
    def transform(self, data):
        """数据转换"""
        raise NotImplementedError
    def load(self, data):
        """数据加载"""
        raise NotImplementedError
    def run(self):
        """运行管道"""
        try:
            data = self.extract()
            transformed = self.transform(data)
            self.load(transformed)
            self.error_handler.succeed()
        except Exception as e:
            self.error_handler.fail(e)
            raise

数据采集层实现

多源数据采集

import pandas as pd
from sqlalchemy import create_engine
import redis
from kafka import KafkaConsumer
import asyncio
import aiohttp
from typing import AsyncGenerator
class DataCollector:
    """多源数据采集器"""
    def __init__(self, config: dict):
        self.config = config
        self.engines = {}
        self.cache = redis.Redis(
            host=config['redis_host'],
            port=config['redis_port'],
            decode_responses=True
        )
    def collect_from_mysql(self, query: str, conn_id: str) -> pd.DataFrame:
        """从MySQL采集数据"""
        if conn_id not in self.engines:
            self.engines[conn_id] = create_engine(
                f"mysql+pymysql://{self.config['mysql_user']}:"
                f"{self.config['mysql_pass']}@{self.config['mysql_host']}/"
                f"{self.config['mysql_db']}?charset=utf8mb4"
            )
        return pd.read_sql(query, self.engines[conn_id])
    async def collect_from_api(self, url: str, params: dict) -> dict:
        """异步采集API数据"""
        async with aiohttp.ClientSession() as session:
            async with session.get(url, params=params) as response:
                return await response.json()
    def collect_from_kafka(self, topic: str, group_id: str):
        """采集Kafka流数据"""
        consumer = KafkaConsumer(
            topic,
            bootstrap_servers=self.config['kafka_servers'],
            group_id=group_id,
            auto_offset_reset='earliest'
        )
        for message in consumer:
            yield message.value

数据质量保障

质量监控和校验

from pydantic import BaseModel, validator
import great_expectations as ge
from datetime import datetime
class DataQualityRule(BaseModel):
    """数据质量规则定义"""
    rule_name: str
    column: str
    rule_type: str  # not_null, unique, range, pattern
    params: dict = {}
    @validator('rule_type')
    def validate_rule_type(cls, v):
        valid_types = ['not_null', 'unique', 'range', 'pattern', 'custom']
        if v not in valid_types:
            raise ValueError(f'Invalid rule type: {v}')
        return v
class DataQualityChecker:
    """数据质量检查器"""
    def __init__(self):
        self.rules = []
        self.quality_metrics = {}
    def add_rule(self, rule: DataQualityRule):
        """添加质量规则"""
        self.rules.append(rule)
    def check_dataframe(self, df: pd.DataFrame) -> dict:
        """检查DataFrame质量"""
        results = {
            'passed': True,
            'violations': [],
            'metrics': {}
        }
        for rule in self.rules:
            if rule.rule_type == 'not_null':
                null_count = df[rule.column].isnull().sum()
                if null_count > 0:
                    results['violations'].append({
                        'rule': rule.rule_name,
                        'column': rule.column,
                        'null_count': null_count
                    })
            elif rule.rule_type == 'unique':
                dup_count = df[rule.column].duplicated().sum()
                if dup_count > 0:
                    results['violations'].append({
                        'rule': rule.rule_name,
                        'column': rule.column,
                        'duplicate_count': dup_count
                    })
        results['passed'] = len(results['violations']) == 0
        results['metrics']['total_rows'] = len(df)
        results['metrics']['timestamp'] = datetime.now()
        return results
    def generate_quality_report(self):
        """生成质量报告"""
        report = {
            'total_checks': len(self.rules),
            'passed_checks': 0,
            'failed_checks': 0,
            'detailed_results': []
        }
        for rule in self.rules:
            result = self.check_rule(rule)
            if result['passed']:
                report['passed_checks'] += 1
            else:
                report['failed_checks'] += 1
            report['detailed_results'].append(result)
        return report

数据处理优化

高性能数据处理

import numpy as np
from concurrent.futures import ThreadPoolExecutor, ProcessPoolExecutor
import dask.dataframe as dd
from functools import lru_cache
class PerformanceOptimizer:
    """性能优化器"""
    def __init__(self):
        self.executor = ProcessPoolExecutor(max_workers=4)
        self.cache_size = 128
    @lru_cache(maxsize=128)
    def get_cached_data(self, query: str):
        """缓存查询结果"""
        return self.execute_query(query)
    def parallel_process(self, data: list, func: callable) -> list:
        """并行处理数据"""
        with ThreadPoolExecutor(max_workers=8) as executor:
            results = list(executor.map(func, data))
        return results
    def chunk_processing(self, df: pd.DataFrame, func: callable, 
                        chunk_size: int = 10000):
        """分块处理大数据"""
        chunks = np.array_split(df, len(df) // chunk_size + 1)
        for chunk in chunks:
            yield func(chunk)
class MemoryOptimizer:
    """内存优化器"""
    @staticmethod
    def optimize_dtypes(df: pd.DataFrame) -> pd.DataFrame:
        """优化数据类型以节省内存"""
        for col in df.columns:
            col_type = df[col].dtype
            if col_type != 'object':
                c_min = df[col].min()
                c_max = df[col].max()
                if str(col_type)[:3] == 'int':
                    if c_min > np.iinfo(np.int8).min and c_max < np.iinfo(np.int8).max:
                        df[col] = df[col].astype(np.int8)
                    elif c_min > np.iinfo(np.int16).min and c_max < np.iinfo(np.int16).max:
                        df[col] = df[col].astype(np.int16)
                    elif c_min > np.iinfo(np.int32).min and c_max < np.iinfo(np.int32).max:
                        df[col] = df[col].astype(np.int32)
                elif str(col_type)[:5] == 'float':
                    if c_min > np.finfo(np.float32).min and c_max < np.finfo(np.float32).max:
                        df[col] = df[col].astype(np.float32)
            else:
                # 对字符串列进行类别编码
                if df[col].nunique() / len(df) < 0.5:
                    df[col] = df[col].astype('category')
        return df

容错与恢复机制

错误处理和恢复

import time
from tenacity import retry, stop_after_attempt, wait_exponential
class ErrorHandler:
    """错误处理器"""
    def __init__(self):
        self.error_log = []
        self.recovery_strategies = {}
    def register_recovery(self, error_type: type, strategy: callable):
        """注册恢复策略"""
        self.recovery_strategies[error_type] = strategy
    def handle_error(self, error: Exception, context: dict = None):
        """处理错误并尝试恢复"""
        error_info = {
            'timestamp': datetime.now(),
            'error_type': type(error).__name__,
            'error_message': str(error),
            'context': context
        }
        self.error_log.append(error_info)
        # 查找恢复策略
        for error_type, strategy in self.recovery_strategies.items():
            if isinstance(error, error_type):
                return strategy(error, context)
        # 默认恢复：重试
        return self.default_recovery(error)
    @retry(stop=stop_after_attempt(3), 
           wait=wait_exponential(multiplier=1, min=4, max=10))
    def default_recovery(self, error):
        """默认恢复策略：重试"""
        raise error
class DataBackup:
    """数据备份恢复"""
    def __init__(self, backup_path: str):
        self.backup_path = backup_path
    def create_snapshot(self, data: pd.DataFrame, snapshot_name: str):
        """创建数据快照"""
        timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
        filepath = f"{self.backup_path}/{snapshot_name}_{timestamp}.parquet"
        data.to_parquet(filepath, compression='snappy')
        return filepath
    def restore_snapshot(self, snapshot_path: str) -> pd.DataFrame:
        """恢复数据快照"""
        return pd.read_parquet(snapshot_path)

监控与告警

系统监控

import psutil
import json
from datetime import datetime, timedelta
class SystemMonitor:
    """系统监控器"""
    def __init__(self, alert_thresholds: dict = None):
        self.thresholds = alert_thresholds or {
            'cpu_percent': 80,
            'memory_percent': 85,
            'disk_percent': 90
        }
        self.metrics_history = []
    def collect_metrics(self):
        """收集系统指标"""
        metrics = {
            'timestamp': datetime.now(),
            'cpu_percent': psutil.cpu_percent(interval=1),
            'memory_percent': psutil.virtual_memory().percent,
            'disk_percent': psutil.disk_usage('/').percent,
            'network_io': psutil.net_io_counters()._asdict()
        }
        self.metrics_history.append(metrics)
        return metrics
    def check_alerts(self):
        """检查是否需要告警"""
        current = self.collect_metrics()
        alerts = []
        for metric, threshold in self.thresholds.items():
            if current[metric] > threshold:
                alerts.append({
                    'metric': metric,
                    'current_value': current[metric],
                    'threshold': threshold,
                    'severity': 'warning' if current[metric] < threshold * 1.2 else 'critical'
                })
        return alerts
    def generate_health_report(self):
        """生成健康报告"""
        metrics = self.collect_metrics()
        alerts = self.check_alerts()
        return {
            'status': 'healthy' if len(alerts) == 0 else 'unhealthy',
            'metrics': metrics,
            'alerts': alerts,
            'recommendations': self._generate_recommendations(metrics)
        }
    def _generate_recommendations(self, metrics: dict) -> list:
        """生成优化建议"""
        recommendations = []
        if metrics['cpu_percent'] > 70:
            recommendations.append("考虑增加计算资源或优化算法")
        if metrics['memory_percent'] > 80:
            recommendations.append("建议清理内存或增加内存容量")
        if metrics['disk_percent'] > 85:
            recommendations.append("建议清理磁盘空间或扩展存储")
        return recommendations

架构最佳实践

完整示例：电商数据体系

class EcommerceDataPipeline:
    """电商数据管道示例"""
    def __init__(self, config: dict):
        self.config = config
        self.collector = DataCollector(config)
        self.quality_checker = DataQualityChecker()
        self.monitor = SystemMonitor()
        self.backup = DataBackup(config['backup_path'])
        # 配置质量规则
        self._setup_quality_rules()
    def _setup_quality_rules(self):
        """设置质量规则"""
        rules = [
            DataQualityRule(
                rule_name="订单ID非空",
                column="order_id",
                rule_type="not_null"
            ),
            DataQualityRule(
                rule_name="用户ID唯一",
                column="user_id",
                rule_type="unique"
            )
        ]
        for rule in rules:
            self.quality_checker.add_rule(rule)
    async def run_daily_pipeline(self):
        """运行每日数据管道"""
        try:
            # 1. 采集数据
            orders = self.collector.collect_from_mysql(
                "SELECT * FROM orders WHERE date = CURRENT_DATE",
                "order_db"
            )
            # 2. 质量检查
            quality_check = self.quality_checker.check_dataframe(orders)
            if not quality_check['passed']:
                self._handle_quality_issues(quality_check)
                return
            # 3. 数据备份
            self.backup.create_snapshot(orders, "orders_daily")
            # 4. 数据转换
            transformed = self._transform_orders(orders)
            # 5. 加载到数据仓库
            self._load_to_warehouse(transformed)
            # 6. 监控和告警
            health = self.monitor.generate_health_report()
            if health['status'] != 'healthy':
                self._send_alerts(health['alerts'])
        except Exception as e:
            self._handle_pipeline_error(e)
    def _transform_orders(self, orders: pd.DataFrame) -> pd.DataFrame:
        """订单数据转换"""
        # 数据清洗和转换逻辑
        orders['order_date'] = pd.to_datetime(orders['order_date'])
        orders['amount_normalized'] = orders['amount'].fillna(0)
        orders['customer_segment'] = orders['total_amount'].apply(
            lambda x: 'high' if x > 1000 else 'medium' if x > 500 else 'low'
        )
        return orders
    def _load_to_warehouse(self, data: pd.DataFrame):
        """加载到数据仓库"""
        engine = create_engine(self.config['warehouse_conn'])
        data.to_sql(
            'fact_orders',
            engine,
            if_exists='append',
            index=False,
            method='multi',
            chunksize=10000
        )

配置管理

import yaml
from pathlib import Path
class ConfigManager:
    """配置管理器"""
    def __init__(self, config_path: str):
        self.config_path = Path(config_path)
        self.config = self._load_config()
    def _load_config(self) -> dict:
        """加载配置"""
        with open(self.config_path, 'r') as f:
            return yaml.safe_load(f)
    def get_database_config(self, db_name: str) -> dict:
        """获取数据库配置"""
        return self.config['databases'].get(db_name, {})
    def get_pipeline_config(self, pipeline_name: str) -> dict:
        """获取管道配置"""
        return self.config['pipelines'].get(pipeline_name, {})

部署与运维

# Docker化部署配置文件示例
"""
version: '3.8'
services:
  data-pipeline:
    build: .
    environment:
      - PYTHONUNBUFFERED=1
      - LOG_LEVEL=INFO
    volumes:
      - ./data:/app/data
      - ./logs:/app/logs
    restart: unless-stopped
    healthcheck:
      test: ["CMD", "python", "-c", "from app import health; health.check()"]
      interval: 30s
      timeout: 10s
      retries: 3
"""