本文目录导读：
我来详细介绍如何用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
"""
- 架构分层：清晰的数据分层（ODS、DWD、DWS、ADS）
- 错误处理：完善的异常捕获和恢复机制
- 性能优化：并行处理、内存优化、缓存策略
- 数据质量：自动化的质量检查和告警
- 监控告警：实时监控系统健康状态
- 备份恢复：定期快照和故障恢复
- 配置管理：外部化配置，环境分离
- 日志记录：全面的日志追踪
这个框架可以根据实际业务需求进行调整和扩展,确保数据体系的稳定性和高效性。
