RailWise KB已发布
批量数据处理模板
RailWise批量数据处理标准化代码模板，支持大规模监测数据的批量清洗、转换、计算、分析与归档，实现高效的大数据量处理工作流。
template-doc
批量数据处理模板
Section titled “批量数据处理模板”1. 模板概述
Section titled “1. 模板概述”1.1 用途
Section titled “1.1 用途”本模板提供大规模监测数据的批量处理标准化流程，支持：
- 批量清洗：缺失值处理、异常值检测、重复数据删除
- 批量转换：格式统一、坐标转换、单位换算
- 批量计算：累计位移计算、变化速率计算、趋势分析
- 批量分析：统计分析、预警检测、报告生成
- 批量归档：数据压缩、分区存储、生命周期管理
1.2 适用场景
Section titled “1.2 适用场景”| 场景 | 数据量 | 处理模式 | 性能要求 |
|---|---|---|---|
| 历史数据迁移 | 百万级 | 离线批处理 | 高吞吐 |
| 多项目汇总 | 十万级 | 并行批处理 | 中等 |
| 日终数据处理 | 万级 | 定时批处理 | 低延迟 |
| 全量数据分析 | 千万级 | 分布式处理 | 高吞吐 |
| 数据归档清理 | 百万级 | 离线批处理 | 高吞吐 |
1.3 输入输出
Section titled “1.3 输入输出”- 输入：原始数据文件、数据库查询结果、数据流
- 输出：清洗后的数据、计算结果、分析报告、归档文件
- 中间产物：处理日志、错误记录、统计报告
2. 代码示例
Section titled “2. 代码示例”2.1 核心批量处理引擎
Section titled “2.1 核心批量处理引擎”"""RailWise 批量数据处理引擎支持大规模监测数据的高效批量处理
作者: RailWise技术部版本: 1.0.0"""
import gcimport loggingimport multiprocessing as mpimport timefrom abc import ABC, abstractmethodfrom concurrent.futures import ProcessPoolExecutor, ThreadPoolExecutorfrom dataclasses import dataclass, fieldfrom datetime import datetime, timedelta, timezonefrom enum import Enumfrom pathlib import Pathfrom typing import Any, Callable, Dict, Iterator, List, Optional, Tuple, Union
import numpy as npimport pandas as pdfrom tqdm import tqdm
# 配置日志logging.basicConfig( level=logging.INFO, format="%(asctime)s - %(name)s - %(levelname)s - %(message)s")logger = logging.getLogger("RailWise.BatchProcessor")
class ProcessingMode(Enum): """处理模式枚举""" SEQUENTIAL = "sequential" # 串行处理 MULTITHREAD = "multithread" # 多线程处理 MULTIPROCESS = "multiprocess" # 多进程处理 DISTRIBUTED = "distributed" # 分布式处理
class DataQualityAction(Enum): """数据质量处理动作""" DROP = "drop" # 删除 FILL_ZERO = "fill_zero" # 填充0 FILL_MEAN = "fill_mean" # 填充均值 FILL_MEDIAN = "fill_median" # 填充中位数 FILL_FORWARD = "fill_forward" # 前向填充 FILL_INTERPOLATE = "interpolate" # 插值填充 FLAG = "flag" # 标记但不处理
@dataclassclass BatchConfig: """批处理配置""" batch_size: int = 10000 # 每批处理记录数 max_workers: int = 4 # 最大工作进程/线程数 processing_mode: ProcessingMode = ProcessingMode.MULTITHREAD memory_limit_mb: int = 1024 # 内存限制(MB) checkpoint_interval: int = 100 # 检查点间隔（批次数） output_dir: Path = field(default_factory=lambda: Path("./batch_output")) temp_dir: Path = field(default_factory=lambda: Path("./batch_temp")) enable_progress_bar: bool = True enable_checkpoint: bool = True retry_count: int = 3 # 失败重试次数 retry_delay_seconds: int = 5 # 重试延迟(秒)
@dataclassclass ProcessingResult: """处理结果""" success: bool records_processed: int = 0 records_failed: int = 0 records_skipped: int = 0 processing_time_ms: float = 0.0 output_files: List[Path] = field(default_factory=list) errors: List[str] = field(default_factory=list) warnings: List[str] = field(default_factory=list) statistics: Dict[str, Any] = field(default_factory=dict)
class DataChunk: """数据分块容器"""
def __init__(self, chunk_id: int, data: pd.DataFrame): self.chunk_id = chunk_id self.data = data self.metadata = { "chunk_id": chunk_id, "record_count": len(data), "columns": list(data.columns), "memory_usage_mb": data.memory_usage(deep=True).sum() / 1024 / 1024 }
def __repr__(self): return f"DataChunk(id={self.chunk_id}, records={self.metadata['record_count']})"
class BatchProcessor(ABC): """批处理抽象基类"""
def __init__(self, config: BatchConfig): self.config = config self.logger = logging.getLogger(self.__class__.__name__) self.checkpoint_data: Dict[str, Any] = {}
# 确保目录存在 self.config.output_dir.mkdir(parents=True, exist_ok=True) self.config.temp_dir.mkdir(parents=True, exist_ok=True)
@abstractmethod def process_chunk(self, chunk: DataChunk) -> DataChunk: """处理单个数据块""" pass
def pre_process(self, data: pd.DataFrame) -> pd.DataFrame: """预处理（可选重写）""" return data
def post_process(self, data: pd.DataFrame) -> pd.DataFrame: """后处理（可选重写）""" return data
def validate_chunk(self, chunk: DataChunk) -> Tuple[bool, List[str]]: """验证数据块""" errors = []
if chunk.data.empty: errors.append(f"数据块 {chunk.chunk_id} 为空")
if chunk.metadata["memory_usage_mb"] > self.config.memory_limit_mb: errors.append(f"数据块 {chunk.chunk_id} 内存使用超限: {chunk.metadata['memory_usage_mb']:.2f}MB")
return len(errors) == 0, errors
def save_checkpoint(self, batch_index: int): """保存检查点""" if not self.config.enable_checkpoint: return
checkpoint_path = self.config.temp_dir / f"checkpoint_{batch_index}.json" import json with open(checkpoint_path, 'w', encoding='utf-8') as f: json.dump(self.checkpoint_data, f, ensure_ascii=False, default=str)
self.logger.info(f"检查点已保存: {checkpoint_path}")
def load_checkpoint(self) -> Optional[Dict[str, Any]]: """加载检查点""" checkpoint_files = sorted(self.config.temp_dir.glob("checkpoint_*.json"))
if checkpoint_files: latest = checkpoint_files[-1] import json with open(latest, 'r', encoding='utf-8') as f: return json.load(f)
return None
def process( self, data_source: Union[pd.DataFrame, Iterator[pd.DataFrame], Path], total_records: Optional[int] = None ) -> ProcessingResult: """ 执行批处理
Args: data_source: 数据源（DataFrame/迭代器/文件路径） total_records: 总记录数（用于进度显示）
Returns: 处理结果 """ start_time = time.time() result = ProcessingResult(success=False)
try: # 1. 加载检查点 checkpoint = self.load_checkpoint() start_batch = checkpoint.get("last_batch", 0) if checkpoint else 0
# 2. 分块处理 chunks = self._create_chunks(data_source)
# 3. 选择处理模式 if self.config.processing_mode == ProcessingMode.SEQUENTIAL: processed_chunks = self._process_sequential(chunks, start_batch, total_records) elif self.config.processing_mode == ProcessingMode.MULTITHREAD: processed_chunks = self._process_multithread(chunks, start_batch, total_records) elif self.config.processing_mode == ProcessingMode.MULTIPROCESS: processed_chunks = self._process_multiprocess(chunks, start_batch, total_records) else: raise NotImplementedError(f"处理模式未实现: {self.config.processing_mode}")
# 4. 合并结果 final_data = self._merge_chunks(processed_chunks) final_data = self.post_process(final_data)
# 5. 保存结果 output_path = self._save_result(final_data) result.output_files.append(output_path)
# 6. 统计 result.records_processed = len(final_data) result.processing_time_ms = (time.time() - start_time) * 1000 result.success = True
self.logger.info( f"批处理完成: {result.records_processed}条记录, " f"耗时: {result.processing_time_ms:.2f}ms" )
except Exception as e: result.errors.append(str(e)) self.logger.error(f"批处理失败: {e}")
return result
def _create_chunks( self, data_source: Union[pd.DataFrame, Iterator[pd.DataFrame], Path] ) -> Iterator[DataChunk]: """创建数据分块""" if isinstance(data_source, pd.DataFrame): # DataFrame直接分块 for i, start in enumerate(range(0, len(data_source), self.config.batch_size)): end = min(start + self.config.batch_size, len(data_source)) chunk = DataChunk(i, data_source.iloc[start:end].copy()) yield chunk
elif isinstance(data_source, Iterator): # 迭代器逐块处理 chunk_idx = 0 for batch_df in data_source: chunk = DataChunk(chunk_idx, batch_df) yield chunk chunk_idx += 1
elif isinstance(data_source, Path): # 文件路径，使用pandas分块读取 if data_source.suffix == '.csv': chunk_idx = 0 for batch_df in pd.read_csv(data_source, chunksize=self.config.batch_size): chunk = DataChunk(chunk_idx, batch_df) yield chunk chunk_idx += 1 elif data_source.suffix in ['.xlsx', '.xls']: # Excel文件需要特殊处理 df = pd.read_excel(data_source) yield from self._create_chunks(df) elif data_source.suffix == '.parquet': # Parquet文件 import pyarrow.parquet as pq parquet_file = pq.ParquetFile(data_source) chunk_idx = 0 for batch in parquet_file.iter_batches(batch_size=self.config.batch_size): df = batch.to_pandas() chunk = DataChunk(chunk_idx, df) yield chunk chunk_idx += 1
def _process_sequential( self, chunks: Iterator[DataChunk], start_batch: int, total_records: Optional[int] ) -> List[DataChunk]: """串行处理""" processed = [] chunk_idx = 0
iterator = tqdm(chunks, total=total_records // self.config.batch_size if total_records else None) \ if self.config.enable_progress_bar else chunks
for chunk in iterator: if chunk_idx < start_batch: chunk_idx += 1 continue
# 验证 valid, errors = self.validate_chunk(chunk) if not valid: self.logger.warning(f"数据块 {chunk.chunk_id} 验证失败: {errors}") continue
# 处理 try: processed_chunk = self.process_chunk(chunk) processed.append(processed_chunk) except Exception as e: self.logger.error(f"处理数据块 {chunk.chunk_id} 失败: {e}")
# 检查点 if (chunk_idx + 1) % self.config.checkpoint_interval == 0: self.save_checkpoint(chunk_idx)
chunk_idx += 1
# 内存管理 if chunk_idx % 10 == 0: gc.collect()
return processed
def _process_multithread( self, chunks: Iterator[DataChunk], start_batch: int, total_records: Optional[int] ) -> List[DataChunk]: """多线程处理""" processed = [] chunk_list = list(chunks) # 先转换为列表
with ThreadPoolExecutor(max_workers=self.config.max_workers) as executor: futures = [] for chunk in chunk_list: if chunk.chunk_id < start_batch: continue future = executor.submit(self._process_chunk_with_retry, chunk) futures.append(future)
# 收集结果 for future in tqdm(futures, disable=not self.config.enable_progress_bar): try: result = future.result() if result is not None: processed.append(result) except Exception as e: self.logger.error(f"多线程处理失败: {e}")
return processed
def _process_multiprocess( self, chunks: Iterator[DataChunk], start_batch: int, total_records: Optional[int] ) -> List[DataChunk]: """多进程处理""" processed = [] chunk_list = list(chunks)
with ProcessPoolExecutor(max_workers=self.config.max_workers) as executor: futures = [] for chunk in chunk_list: if chunk.chunk_id < start_batch: continue future = executor.submit(self._process_chunk_with_retry, chunk) futures.append(future)
for future in tqdm(futures, disable=not self.config.enable_progress_bar): try: result = future.result() if result is not None: processed.append(result) except Exception as e: self.logger.error(f"多进程处理失败: {e}")
return processed
def _process_chunk_with_retry(self, chunk: DataChunk) -> Optional[DataChunk]: """带重试的处理""" for attempt in range(self.config.retry_count): try: return self.process_chunk(chunk) except Exception as e: self.logger.warning(f"处理数据块 {chunk.chunk_id} 第 {attempt + 1} 次尝试失败: {e}") if attempt < self.config.retry_count - 1: time.sleep(self.config.retry_delay_seconds) else: self.logger.error(f"处理数据块 {chunk.chunk_id} 最终失败") return None
def _merge_chunks(self, chunks: List[DataChunk]) -> pd.DataFrame: """合并处理后的数据块""" if not chunks: return pd.DataFrame()
dataframes = [chunk.data for chunk in chunks] return pd.concat(dataframes, ignore_index=True)
def _save_result(self, data: pd.DataFrame) -> Path: """保存处理结果""" timestamp = datetime.now().strftime("%Y%m%d_%H%M%S") output_path = self.config.output_dir / f"batch_result_{timestamp}.parquet"
data.to_parquet(output_path, compression='zstd', index=False) self.logger.info(f"结果已保存: {output_path}")
return output_path
class DataCleaningProcessor(BatchProcessor): """数据清洗处理器"""
def __init__( self, config: BatchConfig, missing_action: DataQualityAction = DataQualityAction.FILL_FORWARD, outlier_method: str = "iqr", outlier_threshold: float = 3.0 ): super().__init__(config) self.missing_action = missing_action self.outlier_method = outlier_method self.outlier_threshold = outlier_threshold
def process_chunk(self, chunk: DataChunk) -> DataChunk: """清洗数据块""" df = chunk.data.copy()
# 1. 处理缺失值 df = self._handle_missing_values(df)
# 2. 处理异常值 df = self._handle_outliers(df)
# 3. 删除重复数据 df = df.drop_duplicates()
# 4. 数据类型转换 df = self._convert_types(df)
return DataChunk(chunk.chunk_id, df)
def _handle_missing_values(self, df: pd.DataFrame) -> pd.DataFrame: """处理缺失值""" numeric_columns = df.select_dtypes(include=[np.number]).columns
if self.missing_action == DataQualityAction.DROP: df = df.dropna() elif self.missing_action == DataQualityAction.FILL_ZERO: df[numeric_columns] = df[numeric_columns].fillna(0) elif self.missing_action == DataQualityAction.FILL_MEAN: df[numeric_columns] = df[numeric_columns].fillna(df[numeric_columns].mean()) elif self.missing_action == DataQualityAction.FILL_MEDIAN: df[numeric_columns] = df[numeric_columns].fillna(df[numeric_columns].median()) elif self.missing_action == DataQualityAction.FILL_FORWARD: df = df.fillna(method='ffill') elif self.missing_action == DataQualityAction.FILL_INTERPOLATE: df[numeric_columns] = df[numeric_columns].interpolate()
return df
def _handle_outliers(self, df: pd.DataFrame) -> pd.DataFrame: """处理异常值""" numeric_columns = df.select_dtypes(include=[np.number]).columns
for col in numeric_columns: if self.outlier_method == "iqr": Q1 = df[col].quantile(0.25) Q3 = df[col].quantile(0.75) IQR = Q3 - Q1 lower = Q1 - self.outlier_threshold * IQR upper = Q3 + self.outlier_threshold * IQR
outlier_mask = (df[col] < lower) | (df[col] > upper) df.loc[outlier_mask, col] = np.nan
elif self.outlier_method == "zscore": mean = df[col].mean() std = df[col].std() z_scores = np.abs((df[col] - mean) / std)
outlier_mask = z_scores > self.outlier_threshold df.loc[outlier_mask, col] = np.nan
return df
def _convert_types(self, df: pd.DataFrame) -> pd.DataFrame: """转换数据类型""" # 时间字段转换 if 'observation_time' in df.columns: df['observation_time'] = pd.to_datetime(df['observation_time'])
# 数值字段优化 for col in df.select_dtypes(include=['float64']).columns: df[col] = df[col].astype('float32')
for col in df.select_dtypes(include=['int64']).columns: df[col] = df[col].astype('int32')
# 分类字段优化 for col in df.select_dtypes(include=['object']).columns: if df[col].nunique() / len(df) < 0.5: df[col] = df[col].astype('category')
return df
class MonitoringCalculationProcessor(BatchProcessor): """监测计算处理器"""
def __init__( self, config: BatchConfig, baseline_date: Optional[datetime] = None, calculate_rate: bool = True, rate_window_hours: int = 24 ): super().__init__(config) self.baseline_date = baseline_date self.calculate_rate = calculate_rate self.rate_window_hours = rate_window_hours
def process_chunk(self, chunk: DataChunk) -> DataChunk: """计算监测指标""" df = chunk.data.copy()
# 确保时间字段 if 'observation_time' in df.columns: df['observation_time'] = pd.to_datetime(df['observation_time']) df = df.sort_values('observation_time')
# 计算累计位移 if 'value_d' in df.columns and self.baseline_date: df = self._calculate_cumulative_displacement(df)
# 计算变化速率 if self.calculate_rate and 'value_d' in df.columns: df = self._calculate_change_rate(df)
# 计算趋势 df = self._calculate_trend(df)
return DataChunk(chunk.chunk_id, df)
def _calculate_cumulative_displacement(self, df: pd.DataFrame) -> pd.DataFrame: """计算累计位移""" if self.baseline_date: baseline = df[df['observation_time'] <= self.baseline_date]['value_d'].iloc[-1] \ if len(df[df['observation_time'] <= self.baseline_date]) > 0 else df['value_d'].iloc[0]
df['cumulative_displacement'] = df['value_d'] - baseline
return df
def _calculate_change_rate(self, df: pd.DataFrame) -> pd.DataFrame: """计算变化速率""" df = df.sort_values('observation_time')
# 使用rolling window计算速率 window = timedelta(hours=self.rate_window_hours)
rates = [] for idx in range(len(df)): current_time = df.iloc[idx]['observation_time'] current_value = df.iloc[idx]['value_d']
# 查找窗口内的数据 window_data = df[ (df['observation_time'] >= current_time - window) & (df['observation_time'] <= current_time) ]
if len(window_data) >= 2: time_diff = (window_data['observation_time'].iloc[-1] - window_data['observation_time'].iloc[0]).total_seconds() / 86400 value_diff = window_data['value_d'].iloc[-1] - window_data['value_d'].iloc[0]
rate = value_diff / time_diff if time_diff > 0 else 0 rates.append(rate) else: rates.append(0)
df['change_rate'] = rates return df
def _calculate_trend(self, df: pd.DataFrame) -> pd.DataFrame: """计算变化趋势""" if 'change_rate' not in df.columns or len(df) < 3: df['trend'] = '稳定' return df
# 使用最近3个点的速率判断趋势 recent_rates = df['change_rate'].tail(3) avg_rate = recent_rates.mean()
if abs(avg_rate) < 0.1: trend = '稳定' elif avg_rate > 0.5: trend = '快速上升' elif avg_rate > 0: trend = '上升' elif avg_rate < -0.5: trend = '快速下降' else: trend = '下降'
df['trend'] = trend return df
class AlertDetectionProcessor(BatchProcessor): """预警检测处理器"""
def __init__( self, config: BatchConfig, thresholds: Dict[str, Dict[str, float]] ): super().__init__(config) self.thresholds = thresholds # {point_type: {yellow: x, orange: y, red: z}}
def process_chunk(self, chunk: DataChunk) -> DataChunk: """检测预警""" df = chunk.data.copy()
# 添加预警等级列 df['alert_level'] = '正常'
for idx, row in df.iterrows(): point_type = row.get('point_type', 'default') cumulative = row.get('cumulative_displacement', row.get('value_d', 0)) rate = row.get('change_rate', 0)
thresholds = self.thresholds.get(point_type, self.thresholds.get('default', {}))
# 判断预警等级 if abs(cumulative) >= thresholds.get('red', float('inf')) or \ abs(rate) >= thresholds.get('red_rate', float('inf')): df.at[idx, 'alert_level'] = '红色预警' elif abs(cumulative) >= thresholds.get('orange', float('inf')) or \ abs(rate) >= thresholds.get('orange_rate', float('inf')): df.at[idx, 'alert_level'] = '橙色预警' elif abs(cumulative) >= thresholds.get('yellow', float('inf')) or \ abs(rate) >= thresholds.get('yellow_rate', float('inf')): df.at[idx, 'alert_level'] = '黄色预警'
return DataChunk(chunk.chunk_id, df)2.2 管道组合处理器
Section titled “2.2 管道组合处理器”"""管道组合处理器 - 将多个处理器串联执行"""
from typing import List
class ProcessingPipeline: """处理管道"""
def __init__(self, processors: List[BatchProcessor]): self.processors = processors self.logger = logging.getLogger("ProcessingPipeline")
def execute( self, data_source: Union[pd.DataFrame, Iterator[pd.DataFrame], Path], total_records: Optional[int] = None ) -> ProcessingResult: """ 执行处理管道
将多个处理器串联执行，前一个的输出作为后一个的输入 """ current_data = data_source final_result = ProcessingResult(success=True)
for i, processor in enumerate(self.processors): self.logger.info(f"执行处理器 {i+1}/{len(self.processors)}: {processor.__class__.__name__}")
result = processor.process(current_data, total_records)
if not result.success: final_result.success = False final_result.errors.extend(result.errors) self.logger.error(f"处理器 {processor.__class__.__name__} 失败") break
# 更新统计 final_result.records_processed = result.records_processed final_result.processing_time_ms += result.processing_time_ms final_result.output_files.extend(result.output_files) final_result.warnings.extend(result.warnings)
# 下一个处理器的输入 if result.output_files: current_data = result.output_files[-1]
self.logger.info(f"处理器 {processor.__class__.__name__} 完成")
return final_result
# 辅助函数：创建标准监测数据处理管道def create_standard_monitoring_pipeline( output_dir: Path, thresholds: Dict[str, Dict[str, float]], baseline_date: Optional[datetime] = None) -> ProcessingPipeline: """ 创建标准监测数据处理管道
处理流程：清洗 -> 计算 -> 预警检测 """ config = BatchConfig( output_dir=output_dir, processing_mode=ProcessingMode.MULTITHREAD, max_workers=4 )
# 1. 数据清洗 cleaning_processor = DataCleaningProcessor( config=config, missing_action=DataQualityAction.FILL_FORWARD, outlier_method="iqr", outlier_threshold=3.0 )
# 2. 监测计算 calculation_processor = MonitoringCalculationProcessor( config=config, baseline_date=baseline_date, calculate_rate=True, rate_window_hours=24 )
# 3. 预警检测 alert_processor = AlertDetectionProcessor( config=config, thresholds=thresholds )
return ProcessingPipeline([ cleaning_processor, calculation_processor, alert_processor ])3. 参数说明
Section titled “3. 参数说明”3.1 BatchConfig 批处理配置参数
Section titled “3.1 BatchConfig 批处理配置参数”| 参数名 | 类型 | 必填 | 默认值 | 说明 |
|---|---|---|---|---|
batch_size |
int | 否 | 10000 | 每批处理记录数 |
max_workers |
int | 否 | 4 | 最大工作进程/线程数 |
processing_mode |
ProcessingMode | 否 | MULTITHREAD | 处理模式 |
memory_limit_mb |
int | 否 | 1024 | 单批内存限制(MB) |
checkpoint_interval |
int | 否 | 100 | 检查点间隔（批次数） |
output_dir |
Path | 否 | ./batch_output |
输出目录 |
temp_dir |
Path | 否 | ./batch_temp |
临时目录 |
enable_progress_bar |
bool | 否 | True | 是否显示进度条 |
enable_checkpoint |
bool | 否 | True | 是否启用检查点 |
retry_count |
int | 否 | 3 | 失败重试次数 |
retry_delay_seconds |
int | 否 | 5 | 重试延迟(秒) |
3.2 DataQualityAction 缺失值处理动作
Section titled “3.2 DataQualityAction 缺失值处理动作”| 动作 | 说明 | 适用场景 |
|---|---|---|
DROP |
删除缺失值行 | 缺失较少 |
FILL_ZERO |
填充0 | 数值型数据 |
FILL_MEAN |
填充均值 | 正态分布数据 |
FILL_MEDIAN |
填充中位数 | 有异常值的数据 |
FILL_FORWARD |
前向填充 | 时间序列数据 |
FILL_INTERPOLATE |
插值填充 | 连续数据 |
FLAG |
标记但不处理 | 需要人工审核 |
4. 使用示例
Section titled “4. 使用示例”4.1 基本批处理流程
Section titled “4.1 基本批处理流程”import pandas as pdfrom datetime import datetime, timezonefrom template_batch_processing import ( BatchConfig, ProcessingMode, DataQualityAction, DataCleaningProcessor, MonitoringCalculationProcessor, AlertDetectionProcessor, ProcessingPipeline)from pathlib import Path
# 1. 准备模拟数据（实际项目中从数据库读取）np.random.seed(42)n_records = 100000
sample_data = pd.DataFrame({ 'record_id': [f'REC_{i:06d}' for i in range(n_records)], 'project_id': ['RW-2024-001'] * n_records, 'point_id': np.random.choice(['JC-01', 'JC-02', 'JC-03', 'JC-04', 'JC-05'], n_records), 'point_type': ['水平位移'] * n_records, 'observation_time': pd.date_range( start='2024-01-01', periods=n_records, freq='H', tz='UTC' ), 'value_d': np.random.normal(0, 2, n_records).cumsum() + np.random.normal(0, 0.5, n_records),})
# 2. 配置批处理config = BatchConfig( batch_size=10000, max_workers=4, processing_mode=ProcessingMode.MULTITHREAD, output_dir=Path("./batch_output"), enable_progress_bar=True)
# 3. 创建处理器# 3.1 数据清洗cleaning_processor = DataCleaningProcessor( config=config, missing_action=DataQualityAction.FILL_FORWARD, outlier_method="iqr", outlier_threshold=3.0)
# 3.2 监测计算calculation_processor = MonitoringCalculationProcessor( config=config, baseline_date=datetime(2024, 1, 1, tzinfo=timezone.utc), calculate_rate=True, rate_window_hours=24)
# 3.3 预警检测thresholds = { '水平位移': { 'yellow': 5.0, 'orange': 8.0, 'red': 10.0, 'yellow_rate': 1.0, 'orange_rate': 2.0, 'red_rate': 3.0 }}
alert_processor = AlertDetectionProcessor( config=config, thresholds=thresholds)
# 4. 创建管道并执行pipeline = ProcessingPipeline([ cleaning_processor, calculation_processor, alert_processor])
result = pipeline.execute(sample_data, total_records=n_records)
# 5. 查看结果print(f"处理成功: {result.success}")print(f"处理记录数: {result.records_processed}")print(f"处理耗时: {result.processing_time_ms:.2f}ms")print(f"输出文件: {result.output_files}")
if result.errors: print(f"错误: {result.errors}")if result.warnings: print(f"警告: {result.warnings}")4.2 从文件批量处理
Section titled “4.2 从文件批量处理”from pathlib import Pathfrom template_batch_processing import BatchConfig, DataCleaningProcessor, DataQualityAction
# 1. 创建大型CSV文件large_file = Path("./large_data.csv")if not large_file.exists(): # 生成测试数据 df = pd.DataFrame({ 'point_id': ['JC-01'] * 1000000, 'observation_time': pd.date_range('2024-01-01', periods=1000000, freq='S'), 'value_d': np.random.normal(0, 1, 1000000).cumsum() }) df.to_csv(large_file, index=False)
# 2. 配置批处理（使用分块读取）config = BatchConfig( batch_size=50000, # 每批5万条 max_workers=4, output_dir=Path("./batch_output"), enable_progress_bar=True)
# 3. 创建处理器processor = DataCleaningProcessor( config=config, missing_action=DataQualityAction.FILL_FORWARD)
# 4. 从文件处理（自动分块读取）result = processor.process(large_file, total_records=1000000)
print(f"处理完成: {result.records_processed}条记录")print(f"输出文件: {result.output_files[0] if result.output_files else '无'}")4.3 使用检查点恢复
Section titled “4.3 使用检查点恢复”from template_batch_processing import BatchConfig, DataCleaningProcessor
# 配置启用检查点config = BatchConfig( batch_size=10000, enable_checkpoint=True, checkpoint_interval=50, # 每50批保存检查点 temp_dir=Path("./batch_temp"))
processor = DataCleaningProcessor(config=config)
# 模拟处理中断后恢复# 第一次运行：处理到第150批时中断# 第二次运行：自动从检查点恢复result = processor.process(large_file, total_records=1000000)
# 检查点文件会自动加载，从上次中断处继续print(f"从检查点恢复后处理完成")4.4 自定义处理器
Section titled “4.4 自定义处理器”from template_batch_processing import BatchProcessor, DataChunk, BatchConfig
class CustomStatisticsProcessor(BatchProcessor): """自定义统计处理器"""
def __init__(self, config: BatchConfig, group_by: str = 'point_id'): super().__init__(config) self.group_by = group_by self.statistics = []
def process_chunk(self, chunk: DataChunk) -> DataChunk: """计算分组统计""" df = chunk.data
# 按测点分组统计 stats = df.groupby(self.group_by).agg({ 'value_d': ['mean', 'std', 'min', 'max', 'count'], 'change_rate': ['mean', 'max'] }).reset_index()
# 展平列名 stats.columns = ['_'.join(col).strip('_') for col in stats.columns.values]
self.statistics.append(stats)
return chunk # 返回原始数据，不修改
def get_statistics(self) -> pd.DataFrame: """获取合并后的统计结果""" if not self.statistics: return pd.DataFrame() return pd.concat(self.statistics, ignore_index=True)
# 使用自定义处理器config = BatchConfig(output_dir=Path("./stats_output"))stats_processor = CustomStatisticsProcessor(config, group_by='point_id')
result = stats_processor.process(sample_data)statistics = stats_processor.get_statistics()
print("分组统计结果:")print(statistics.head())5. 输出示例
Section titled “5. 输出示例”5.1 处理结果统计
Section titled “5.1 处理结果统计”处理成功: True处理记录数: 100000处理耗时: 3456.78ms输出文件: [PosixPath('batch_output/batch_result_20240708_143052.parquet')]
警告: []5.2 清洗后数据示例
Section titled “5.2 清洗后数据示例”# 读取处理结果df = pd.read_parquet("batch_output/batch_result_20240708_143052.parquet")
print(df.head())输出：
record_id project_id point_id ... value_d cumulative_displacement change_rate alert_level0 REC_000000 RW-2024-001 JC-01 ... 0.12 0.12 0.00 正常1 REC_000001 RW-2024-001 JC-01 ... 0.15 0.27 0.15 正常2 REC_000002 RW-2024-001 JC-01 ... 0.08 0.35 0.08 正常3 REC_000003 RW-2024-001 JC-01 ... 1.85 2.20 1.85 黄色预警4 REC_000004 RW-2024-001 JC-01 ... 0.03 2.23 0.03 正常5.3 检查点文件
Section titled “5.3 检查点文件”{ "last_batch": 150, "processed_records": 1500000, "timestamp": "2024-07-08T14:30:52.123456", "processor_state": { "cleaning_stats": { "missing_values_filled": 234, "outliers_detected": 12 } }}6. 扩展指南
Section titled “6. 扩展指南”6.1 添加自定义处理器
Section titled “6.1 添加自定义处理器”from template_batch_processing import BatchProcessor, DataChunk, BatchConfig
class DataValidationProcessor(BatchProcessor): """自定义数据验证处理器"""
def __init__(self, config: BatchConfig, rules: List[Dict]): super().__init__(config) self.rules = rules self.validation_errors = []
def process_chunk(self, chunk: DataChunk) -> DataChunk: """验证数据块""" df = chunk.data
for rule in self.rules: column = rule['column'] condition = rule['condition'] message = rule['message']
# 应用验证规则 invalid_mask = ~df.eval(condition) invalid_count = invalid_mask.sum()
if invalid_count > 0: self.validation_errors.append({ 'chunk_id': chunk.chunk_id, 'rule': message, 'invalid_count': int(invalid_count) })
# 标记无效数据 df.loc[invalid_mask, 'validation_status'] = 'INVALID'
return DataChunk(chunk.chunk_id, df)
def get_validation_report(self) -> pd.DataFrame: """获取验证报告""" return pd.DataFrame(self.validation_errors)
# 使用rules = [ {'column': 'value_d', 'condition': 'value_d >= -100 and value_d <= 100', 'message': '位移值超出合理范围'}, {'column': 'observation_time', 'condition': 'observation_time <= @pd.Timestamp.now()', 'message': '观测时间在未来'}]
validator = DataValidationProcessor(config, rules)result = validator.process(data)report = validator.get_validation_report()print(report)6.2 分布式处理扩展
Section titled “6.2 分布式处理扩展”# 使用Dask进行分布式处理import dask.dataframe as dd
class DaskBatchProcessor(BatchProcessor): """基于Dask的分布式处理器"""
def process(self, data_source, total_records=None): """使用Dask执行分布式处理""" if isinstance(data_source, pd.DataFrame): ddf = dd.from_pandas(data_source, npartitions=self.config.max_workers) elif isinstance(data_source, Path): ddf = dd.read_parquet(data_source) else: raise ValueError("不支持的数据源类型")
# 应用处理函数 result = ddf.map_partitions(self._process_partition)
# 计算并保存 output_path = self.config.output_dir / "dask_result.parquet" result.to_parquet(output_path, compression='zstd')
return ProcessingResult( success=True, records_processed=len(result), output_files=[output_path] )
def _process_partition(self, df): """处理单个分区""" chunk = DataChunk(0, df) processed = self.process_chunk(chunk) return processed.data6.3 集成到定时任务
Section titled “6.3 集成到定时任务”from apscheduler.schedulers.background import BackgroundSchedulerfrom datetime import datetime
def daily_batch_process(): """每日批处理任务""" logger.info("开始每日批处理")
# 1. 从数据库读取昨日数据 # data = query_yesterday_data()
# 2. 创建管道 pipeline = create_standard_monitoring_pipeline( output_dir=Path(f"./batch_output/{datetime.now().strftime('%Y%m%d')}"), thresholds=thresholds, baseline_date=datetime.now() - timedelta(days=30) )
# 3. 执行处理 # result = pipeline.execute(data)
logger.info("每日批处理完成")
# 每日凌晨2点执行scheduler = BackgroundScheduler()scheduler.add_job(daily_batch_process, 'cron', hour=2, minute=0)scheduler.start()6.4 内存优化策略
Section titled “6.4 内存优化策略”class MemoryOptimizedProcessor(BatchProcessor): """内存优化处理器"""
def process_chunk(self, chunk: DataChunk) -> DataChunk: """内存优化的处理""" df = chunk.data
# 1. 选择需要的列 needed_columns = ['point_id', 'observation_time', 'value_d', 'change_rate'] df = df[needed_columns]
# 2. 类型优化 df['point_id'] = df['point_id'].astype('category') df['value_d'] = df['value_d'].astype('float32')
# 3. 处理完成后立即释放内存 import gc gc.collect()
return DataChunk(chunk.chunk_id, df)
def _merge_chunks(self, chunks: List[DataChunk]) -> pd.DataFrame: """增量合并，避免一次性加载所有数据""" output_path = self.config.output_dir / "merged_result.parquet"
for i, chunk in enumerate(chunks): if i == 0: chunk.data.to_parquet(output_path, index=False) else: # 追加写入 existing = pd.read_parquet(output_path) merged = pd.concat([existing, chunk.data], ignore_index=True) merged.to_parquet(output_path, index=False, compression='zstd')
# 释放内存 del existing, merged gc.collect()
return pd.read_parquet(output_path)7. 常见问题
Section titled “7. 常见问题”Q1: 处理大数据量时内存溢出怎么办？
Section titled “Q1: 处理大数据量时内存溢出怎么办？”A: 采用以下策略：
- 减小
batch_size - 使用
MULTIPROCESS模式 - 启用分块读取（从文件）
- 优化数据类型（float64→float32）
- 只选择需要的列
config = BatchConfig( batch_size=5000, # 减小批次 processing_mode=ProcessingMode.MULTIPROCESS, memory_limit_mb=512 # 限制内存)Q2: 如何处理处理中断后的恢复？
Section titled “Q2: 如何处理处理中断后的恢复？”A: 启用检查点功能：
config = BatchConfig( enable_checkpoint=True, checkpoint_interval=50, # 每50批保存检查点 temp_dir=Path("./checkpoints"))Q3: 多进程模式下数据共享问题？
Section titled “Q3: 多进程模式下数据共享问题？”A: 多进程模式下每个进程有独立内存空间，避免共享状态：
# 错误：多进程共享状态class BadProcessor(BatchProcessor): shared_list = [] # 不要在类级别共享状态
# 正确：每个进程独立处理class GoodProcessor(BatchProcessor): def __init__(self, config): super().__init__(config) self.local_results = [] # 实例级别状态Q4: 如何监控处理进度？
Section titled “Q4: 如何监控处理进度？”A: 使用tqdm进度条和日志：
config = BatchConfig( enable_progress_bar=True # 显示进度条)
# 自定义进度回调from tqdm import tqdm
with tqdm(total=total_records, desc="处理中") as pbar: for chunk in chunks: process_chunk(chunk) pbar.update(len(chunk.data))Q5: 如何优化处理速度？
Section titled “Q5: 如何优化处理速度？”A:
- 使用Parquet格式（比CSV快10-100倍）
- 使用向量化操作（避免循环）
- 使用Numba加速计算
- 合理设置并行度（通常CPU核心数-1）
import numba
@numba.jit(nopython=True)def fast_calculation(values): """Numba加速计算""" result = np.empty_like(values) for i in range(len(values)): result[i] = values[i] * 2.0 return result相关文档链接
Section titled “相关文档链接”- 监测数据导入标准化模板 — 数据导入
- 数据导出与格式转换模板 — 处理结果导出
- 监测报告自动生成模板 — 基于处理结果生成报告
- API接口集成模板 — 通过API获取数据
ai_tags: domain: ["工程监测", "数据处理", "大数据"] technology: ["Python", "Pandas", "NumPy", "多线程", "多进程"] standard: ["GB 50497", "CJJ/T 202", "JGJ 8"] product: ["RAILWISE-TSM", "RAILWISE-CLI", "WorkWise"] difficulty: "advanced" role: ["数据工程师", "监测工程师", "系统架构师"] task_type: ["批处理", "数据清洗", "性能优化"]