未记录事实核验 · 状态：维护中
记录仅说明核验时间与维护状态，不保证适用于所有环境；使用前请留意正文中的版本、适用条件与风险提示。
warning 适用边界与维护说明
本文适合在隔离测试目录中学习文件读写、格式转换、分块处理、数据库操作、性能对比和ETL组织，不应直接作为生产级并发存储或业务质量准入框架；示例会覆盖并删除固定名称的测试文件。各代码块依赖前序导入，运行需准备相应第三方库，fcntl要求Unix环境，Parquet输出需要pyarrow或fastparquet。当前CSV跨块均值未按非缺失值数量加权，缺失计数未完整累加；文件锁删除及共享锁写入存在互斥风险，原地覆盖不保证原子替换，锁超时和死锁检测未实现；SQLite自动提交配置下不能据此保证批次整体回滚或共享连接并发安全。大型Excel处理会累计全部结果，mmap切片仍复制数据，不能将示例理解为内存有界处理或完整零拷贝实现。ETL数据库提取为占位、数据库加载明确未实现，自定义质量检查失败不会自动阻断加载，valid_records仅扣除重复记录，后续转换阶段的输入行数血缘可能误记；自动重试和增量更新也未交付。CSV类所列自适应内存管理、并行读写及pyarrow支持并未由当前示例实现，写入计时也不是不同解析引擎的写入性能比较。性能输出与总结中的优化建议需结合数据、库版本和运行环境判断，不构成通用提速或可靠性保证；本次仅作全文结构审阅，未运行代码或作外部事实核验。
Python文件操作与数据I/O系统
abstract 学习目标与内容
核心能力：掌握Python文件I/O底层机制和高效数据处理技术 技术要点：文件系统操作、缓冲策略、流式处理、并发访问控制 实践应用：大文件处理、数据管道构建、性能优化实战 前置知识：Python基础语法、Python数据结构
文件I/O系统架构
底层机制与原理
import os
import sys
import io
from pathlib import Path
import mmap
import fcntl
import threading
import asyncio
from concurrent.futures import ThreadPoolExecutor, ProcessPoolExecutor
import time
import numpy as np
import pandas as pd
print("=== Python文件I/O系统底层机制 ===")
# 1. 文件描述符和系统调用层
class FileIODemo:
def __init__(self):
self.buffer_sizes = [4096, 8192, 16384, 32768, 65536]
self.io_modes = ['text', 'binary', 'buffered', 'unbuffered']
def demonstrate_file_descriptors(self):
"""演示文件描述符的工作机制"""
print("\n1. 文件描述符层次结构:")
# 创建测试文件
test_file = "fd_test.txt"
with open(test_file, 'w') as f:
# 获取文件描述符
fd = f.fileno()
print(f"文件描述符: {fd}")
print(f"文件对象: {f}")
print(f"文件模式: {f.mode}")
print(f"文件编码: {f.encoding}")
# 系统级文件信息
stat_info = os.fstat(fd)
print(f"文件大小: {stat_info.st_size} bytes")
print(f"最后修改: {stat_info.st_mtime}")
print(f"inode: {stat_info.st_ino}")
# 清理文件
if os.path.exists(test_file):
os.remove(test_file)
def demonstrate_buffer_strategies(self):
"""演示不同缓冲策略的性能差异"""
print("\n2. 缓冲策略性能分析:")
test_data = "test data line\n" * 10000
for buffer_size in self.buffer_sizes:
start_time = time.time()
with open(f"buffer_test_{buffer_size}.txt", 'w', buffering=buffer_size) as f:
f.write(test_data)
write_time = time.time() - start_time
# 读取测试
start_time = time.time()
with open(f"buffer_test_{buffer_size}.txt", 'r', buffering=buffer_size) as f:
content = f.read()
read_time = time.time() - start_time
file_size = os.path.getsize(f"buffer_test_{buffer_size}.txt")
print(f"缓冲区 {buffer_size:6d}: 写入 {write_time:.4f}s, 读取 {read_time:.4f}s, 大小 {file_size} bytes")
# 清理测试文件
os.remove(f"buffer_test_{buffer_size}.txt")
# 实例化并运行演示
demo = FileIODemo()
demo.demonstrate_file_descriptors()
demo.demonstrate_buffer_strategies()
高性能文件处理技术
CSV文件处理：解析引擎与优化策略
import pandas as pd
import numpy as np
import csv
import io
from pathlib import Path
import time
from typing import Iterator, Dict, Any, Optional
print("=== 高性能CSV文件处理系统 ===")
class CSVProcessor:
"""高性能CSV处理器
技术特性:
- 多引擎支持 (c, python, pyarrow)
- 自适应内存管理
- 流式处理能力
- 并行读写优化
"""
def __init__(self):
self.engines = ['c', 'python'] # 可用解析引擎
self.chunk_strategies = ['memory_based', 'row_based', 'adaptive']
def create_test_data(self, size: int = 100000) -> pd.DataFrame:
"""创建测试数据集"""
print(f"\n生成 {size:,} 行测试数据...")
np.random.seed(42)
data = {
'id': range(size),
'timestamp': pd.date_range('2024-01-01', periods=size, freq='1min'),
'category': np.random.choice(['A', 'B', 'C', 'D', 'E'], size),
'value': np.random.randn(size) * 100,
'status': np.random.choice([0, 1, 2], size),
'description': [f'record_{i}_with_details' for i in range(size)]
}
df = pd.DataFrame(data)
# 引入数据质量问题
missing_indices = np.random.choice(size, size // 20, replace=False)
df.loc[missing_indices, 'value'] = np.nan
return df
def benchmark_engines(self, df: pd.DataFrame) -> Dict[str, Dict[str, float]]:
"""对比不同解析引擎的性能"""
print("\n=== 解析引擎性能基准测试 ===")
results = {}
test_file = 'engine_test.csv'
# 保存测试文件
df.to_csv(test_file, index=False)
file_size = Path(test_file).stat().st_size / 1024 / 1024 # MB
for engine in self.engines:
print(f"\n测试引擎: {engine}")
# 写入性能测试
start_time = time.time()
df.to_csv(f'{engine}_write_test.csv', index=False)
write_time = time.time() - start_time
# 读取性能测试
start_time = time.time()
try:
read_df = pd.read_csv(test_file, engine=engine)
read_time = time.time() - start_time
success = True
except Exception as e:
read_time = float('inf')
success = False
print(f"引擎 {engine} 失败: {e}")
results[engine] = {
'write_time': write_time,
'read_time': read_time,
'file_size_mb': file_size,
'success': success
}
print(f"写入: {write_time:.3f}s, 读取: {read_time:.3f}s")
# 清理临时文件
for temp_file in [f'{engine}_write_test.csv']:
if Path(temp_file).exists():
Path(temp_file).unlink()
# 清理主测试文件
if Path(test_file).exists():
Path(test_file).unlink()
return results
def demonstrate_streaming(self, file_path: str, chunk_size: int = 10000):
"""演示流式处理大文件"""
print(f"\n=== 流式处理演示 (chunk_size: {chunk_size:,}) ===")
total_rows = 0
chunk_count = 0
aggregated_stats = {}
try:
start_time = time.time()
# 使用生成器进行流式处理
for chunk in pd.read_csv(file_path, chunksize=chunk_size):
chunk_count += 1
total_rows += len(chunk)
# 对每个chunk进行统计
chunk_stats = {
'mean_value': chunk['value'].mean() if 'value' in chunk else 0,
'category_counts': chunk['category'].value_counts().to_dict() if 'category' in chunk else {},
'missing_count': chunk.isnull().sum().sum()
}
# 累积统计
if not aggregated_stats:
aggregated_stats = chunk_stats
else:
# 合并统计信息
aggregated_stats['mean_value'] = (
(aggregated_stats['mean_value'] * (chunk_count - 1) + chunk_stats['mean_value']) / chunk_count
)
for cat, count in chunk_stats['category_counts'].items():
aggregated_stats['category_counts'][cat] = (
aggregated_stats['category_counts'].get(cat, 0) + count
)
if chunk_count % 10 == 0:
print(f"已处理 {chunk_count} 个chunk, {total_rows:,} 行数据")
processing_time = time.time() - start_time
print(f"\n流式处理完成:")
print(f"- 总chunk数: {chunk_count}")
print(f"- 总行数: {total_rows:,}")
print(f"- 处理时间: {processing_time:.3f}s")
print(f"- 处理速度: {total_rows/processing_time:,.0f} 行/秒")
print(f"- 平均值: {aggregated_stats['mean_value']:.2f}")
print(f"- 类别分布: {aggregated_stats['category_counts']}")
except Exception as e:
print(f"流式处理错误: {e}")
# 实例化处理器并运行测试
processor = CSVProcessor()
test_data = processor.create_test_data(50000)
# 保存测试文件
test_csv_file = 'large_test.csv'
test_data.to_csv(test_csv_file, index=False)
# 运行基准测试
engine_results = processor.benchmark_engines(test_data.head(1000)) # 使用小样本进行引擎测试
# 演示流式处理
processor.demonstrate_streaming(test_csv_file, chunk_size=5000)
# 清理测试文件
if Path(test_csv_file).exists():
Path(test_csv_file).unlink()
print(f"\n已清理测试文件: {test_csv_file}")
结构化文件处理：Excel与复杂格式
from openpyxl import Workbook, load_workbook
from openpyxl.styles import Font, PatternFill, Alignment
from openpyxl.utils.dataframe import dataframe_to_rows
from openpyxl.chart import LineChart, Reference
import xlsxwriter
from typing import List, Dict
print("=== 高级Excel文件处理系统 ===")
class ExcelProcessor:
"""Excel文件高级处理器
功能特性:
- 多引擎支持 (openpyxl, xlsxwriter, xlrd)
- 样式和格式化控制
- 图表和可视化集成
- 大文件优化处理
- 数据验证和完整性检查
"""
def __init__(self):
self.supported_engines = ['openpyxl', 'xlsxwriter']
self.max_excel_rows = 1048576 # Excel行限制
self.max_excel_cols = 16384 # Excel列限制
def create_advanced_excel(self, data_dict: Dict[str, pd.DataFrame], filename: str):
"""创建高级Excel文件，包含样式和图表"""
print(f"\n创建高级Excel文件: {filename}")
with pd.ExcelWriter(filename, engine='xlsxwriter') as writer:
workbook = writer.book
# 定义样式
header_format = workbook.add_format({
'bold': True,
'text_wrap': True,
'valign': 'top',
'fg_color': '#4472C4',
'font_color': 'white',
'border': 1
})
number_format = workbook.add_format({'num_format': '#,##0.00'})
percent_format = workbook.add_format({'num_format': '0.00%'})
date_format = workbook.add_format({'num_format': 'yyyy-mm-dd'})
for sheet_name, df in data_dict.items():
# 检查数据大小限制
if len(df) > self.max_excel_rows - 1:
print(f"警告: {sheet_name} 数据行数 ({len(df)}) 超过扣除表头后的Excel限制，将截断")
df = df.head(self.max_excel_rows - 1)
# 写入数据
df.to_excel(writer, sheet_name=sheet_name, index=False, header=False, startrow=1)
worksheet = writer.sheets[sheet_name]
# 应用表头样式
for col_num, value in enumerate(df.columns.values):
worksheet.write(0, col_num, value, header_format)
# 自动调整列宽
for idx, col in enumerate(df.columns):
max_length = max(
df[col].astype(str).str.len().max(),
len(str(col))
)
worksheet.set_column(idx, idx, min(max_length + 2, 50))
# 应用数字格式
for idx, col in enumerate(df.columns):
if df[col].dtype in ['float64', 'int64']:
if col.lower().find('percent') != -1 or col.lower().find('rate') != -1:
worksheet.set_column(idx, idx, None, percent_format)
else:
worksheet.set_column(idx, idx, None, number_format)
elif df[col].dtype == 'datetime64[ns]':
worksheet.set_column(idx, idx, None, date_format)
# 添加数据筛选
worksheet.autofilter(0, 0, len(df), len(df.columns) - 1)
# 为数值数据创建图表
numeric_cols = df.select_dtypes(include=[np.number]).columns
if len(numeric_cols) > 0 and len(df) > 1:
chart = workbook.add_chart({'type': 'line'})
for col_idx, col_name in enumerate(numeric_cols[:3]): # 最多3个系列
chart.add_series({
'name': col_name,
'categories': [sheet_name, 1, 0, len(df), 0],
'values': [sheet_name, 1, df.columns.get_loc(col_name),
len(df), df.columns.get_loc(col_name)],
})
chart.set_title({'name': f'{sheet_name} 趋势图'})
chart.set_x_axis({'name': '数据点'})
chart.set_y_axis({'name': '数值'})
# 插入图表
worksheet.insert_chart('H2', chart)
print(f"✅ 高级Excel文件创建完成: {filename}")
return filename
def read_excel_optimized(self, filename: str, **kwargs) -> Dict[str, pd.DataFrame]:
"""优化的Excel读取方法"""
print(f"\n优化读取Excel文件: {filename}")
start_time = time.time()
# 首先获取工作表信息
excel_file = pd.ExcelFile(filename)
sheet_names = excel_file.sheet_names
print(f"发现 {len(sheet_names)} 个工作表: {sheet_names}")
dataframes = {}
for sheet_name in sheet_names:
try:
# 使用优化参数读取
df = pd.read_excel(
filename,
sheet_name=sheet_name,
engine='openpyxl', # 推荐引擎
**kwargs
)
dataframes[sheet_name] = df
print(f"- {sheet_name}: {df.shape[0]} 行 × {df.shape[1]} 列")
except Exception as e:
print(f"❌ 读取工作表 {sheet_name} 失败: {e}")
read_time = time.time() - start_time
print(f"总读取时间: {read_time:.3f}s")
return dataframes
def process_large_excel(self, filename: str, max_rows: int = 100000):
"""处理大型Excel文件的策略"""
print(f"\n处理大型Excel文件: {filename}")
try:
# 分块读取策略
chunk_size = 10000
all_chunks = []
# 使用openpyxl直接操作
wb = load_workbook(filename, read_only=True)
for sheet_name in wb.sheetnames:
ws = wb[sheet_name]
print(f"处理工作表: {sheet_name}")
# 获取数据范围
max_row = ws.max_row
max_col = ws.max_column
print(f"- 数据范围: {max_row} 行 × {max_col} 列")
if max_row > max_rows:
print(f"- 文件较大，使用分块处理策略")
# 分块处理
for start_row in range(2, max_row + 1, chunk_size): # 跳过标题行
end_row = min(start_row + chunk_size - 1, max_row)
# 读取数据块
chunk_data = []
for row in ws.iter_rows(min_row=start_row, max_row=end_row,
values_only=True):
chunk_data.append(row)
if chunk_data:
# 获取列名
headers = [cell.value for cell in ws[1]]
chunk_df = pd.DataFrame(chunk_data, columns=headers)
# 进行必要的处理
processed_chunk = self._process_chunk(chunk_df)
all_chunks.append(processed_chunk)
print(f" 已处理行 {start_row}-{end_row}")
else:
# 直接读取小文件
df = pd.read_excel(filename, sheet_name=sheet_name)
all_chunks.append(self._process_chunk(df))
wb.close()
# 合并所有块
if all_chunks:
final_df = pd.concat(all_chunks, ignore_index=True)
print(f"\n合并完成: {len(final_df)} 行数据")
return final_df
except Exception as e:
print(f"❌ 处理大文件失败: {e}")
return None
def _process_chunk(self, chunk: pd.DataFrame) -> pd.DataFrame:
"""处理数据块的通用方法"""
# 基本数据清理
chunk = chunk.dropna(how='all') # 删除全空行
# 数据类型优化
for col in chunk.select_dtypes(include=['object']):
if chunk[col].nunique() / len(chunk) < 0.5: # 如果唯一值比例小于50%
chunk[col] = chunk[col].astype('category')
return chunk
# 创建示例数据
sample_data = {
'销售数据': pd.DataFrame({
'日期': pd.date_range('2024-01-01', periods=100),
'销售额': np.random.randint(1000, 10000, 100),
'订单数': np.random.randint(10, 100, 100),
'转化率': np.random.uniform(0.01, 0.1, 100)
}),
'产品分析': pd.DataFrame({
'产品ID': [f'P{i:03d}' for i in range(50)],
'产品名称': [f'产品{i}' for i in range(50)],
'价格': np.random.uniform(10, 1000, 50),
'库存': np.random.randint(0, 500, 50)
})
}
# 实例化处理器
excel_processor = ExcelProcessor()
# 创建高级Excel文件
advanced_excel_file = 'advanced_report.xlsx'
excel_processor.create_advanced_excel(sample_data, advanced_excel_file)
# 优化读取
read_data = excel_processor.read_excel_optimized(advanced_excel_file)
# 显示读取结果
for sheet_name, df in read_data.items():
print(f"\n{sheet_name} 数据预览:")
print(df.head(3))
# 清理文件
if Path(advanced_excel_file).exists():
Path(advanced_excel_file).unlink()
print(f"\n已清理文件: {advanced_excel_file}")
数据库连接与持久化存储
SQLite数据库：轻量级本地存储解决方案
import sqlite3
from sqlalchemy import create_engine, text, MetaData, Table, inspect
from sqlalchemy.pool import StaticPool
from contextlib import contextmanager
import threading
from typing import Generator, List, Dict, Any, Optional
print("=== 数据库连接与事务管理系统 ===")
class DatabaseManager:
"""高性能数据库管理器
技术特性:
- 连接池管理
- 事务控制
- 批量操作优化
- 共享连接配置示例，不保证并发安全
- 尚未实现自动重试机制
"""
def __init__(self, db_url: str = 'sqlite:///advanced_demo.db'):
self.db_url = db_url
self.engine = None
self._local = threading.local()
self._setup_engine()
def _setup_engine(self):
"""配置数据库引擎"""
print(f"\n初始化数据库连接: {self.db_url}")
# SQLite连接配置
if 'sqlite' in self.db_url:
self.engine = create_engine(
self.db_url,
poolclass=StaticPool,
pool_pre_ping=True,
connect_args={
'check_same_thread': False,
'timeout': 30,
'isolation_level': None # 启用autocommit模式
},
echo=False # 设置为True可看到SQL语句
)
else:
# 其他数据库连接配置
self.engine = create_engine(
self.db_url,
pool_size=10,
max_overflow=20,
pool_pre_ping=True,
pool_recycle=3600
)
print(f"✅ 数据库引擎配置完成")
@contextmanager
def get_connection(self):
"""获取数据库连接的上下文管理器"""
conn = self.engine.connect()
try:
yield conn
finally:
conn.close()
@contextmanager
def transaction(self):
"""事务上下文管理器"""
with self.get_connection() as conn:
trans = conn.begin()
try:
yield conn
trans.commit()
except Exception:
trans.rollback()
raise
def create_sample_data(self) -> Dict[str, pd.DataFrame]:
"""创建示例数据"""
print("\n创建示例数据集...")
# 用户表
users_df = pd.DataFrame({
'user_id': range(1, 1001),
'username': [f'user_{i:04d}' for i in range(1, 1001)],
'email': [f'user{i}@example.com' for i in range(1, 1001)],
'created_at': pd.date_range('2023-01-01', periods=1000, freq='H'),
'status': np.random.choice(['active', 'inactive', 'suspended'], 1000),
'score': np.random.randint(0, 1000, 1000)
})
# 订单表
orders_df = pd.DataFrame({
'order_id': range(1, 5001),
'user_id': np.random.choice(range(1, 1001), 5000),
'product_id': np.random.choice(range(1, 201), 5000),
'quantity': np.random.randint(1, 10, 5000),
'price': np.random.uniform(10.0, 500.0, 5000),
'order_date': pd.date_range('2024-01-01', periods=5000, freq='30min'),
'status': np.random.choice(['pending', 'completed', 'cancelled'], 5000)
})
# 产品表
products_df = pd.DataFrame({
'product_id': range(1, 201),
'product_name': [f'产品{i}' for i in range(1, 201)],
'category': np.random.choice(['电子产品', '服装', '书籍', '家居', '运动'], 200),
'price': np.random.uniform(5.0, 1000.0, 200),
'stock': np.random.randint(0, 1000, 200)
})
return {
'users': users_df,
'orders': orders_df,
'products': products_df
}
def bulk_insert_optimized(self, table_name: str, df: pd.DataFrame,
batch_size: int = 1000) -> int:
"""优化的批量插入方法"""
print(f"\n批量插入 {table_name}: {len(df):,} 行数据")
start_time = time.time()
total_inserted = 0
try:
with self.transaction() as conn:
# 分批插入
for i in range(0, len(df), batch_size):
batch = df.iloc[i:i + batch_size]
# 使用pandas的to_sql方法
batch.to_sql(
table_name,
conn,
if_exists='append',
index=False,
method='multi' # 使用批量插入
)
total_inserted += len(batch)
if i % (batch_size * 10) == 0:
print(f" 已插入: {total_inserted:,} 行")
insert_time = time.time() - start_time
print(f"✅ 批量插入完成: {total_inserted:,} 行，耗时 {insert_time:.3f}s")
print(f" 插入速度: {total_inserted/insert_time:,.0f} 行/秒")
return total_inserted
except Exception as e:
print(f"❌ 批量插入失败: {e}")
return 0
def execute_complex_query(self, query: str, params: Dict[str, Any] = None) -> pd.DataFrame:
"""执行复杂查询"""
print(f"\n执行复杂查询...")
start_time = time.time()
try:
with self.get_connection() as conn:
if params:
result_df = pd.read_sql(text(query), conn, params=params)
else:
result_df = pd.read_sql(query, conn)
query_time = time.time() - start_time
print(f"✅ 查询完成: 返回 {len(result_df):,} 行，耗时 {query_time:.3f}s")
return result_df
except Exception as e:
print(f"❌ 查询执行失败: {e}")
return pd.DataFrame()
def create_indexes(self, index_definitions: List[Dict[str, Any]]):
"""创建数据库索引"""
print(f"\n创建数据库索引...")
with self.get_connection() as conn:
for index_def in index_definitions:
try:
index_sql = f"CREATE INDEX IF NOT EXISTS {index_def['name']} ON {index_def['table']} ({index_def['columns']})"
conn.execute(text(index_sql))
print(f"✅ 索引创建成功: {index_def['name']}")
except Exception as e:
print(f"❌ 索引创建失败 {index_def['name']}: {e}")
def analyze_performance(self, query: str) -> Dict[str, Any]:
"""分析查询性能"""
print(f"\n分析查询性能...")
with self.get_connection() as conn:
# SQLite查询计划
explain_query = f"EXPLAIN QUERY PLAN {query}"
try:
plan_df = pd.read_sql(explain_query, conn)
print("查询执行计划:")
print(plan_df)
# 执行查询并计时
start_time = time.time()
result_df = pd.read_sql(query, conn)
execution_time = time.time() - start_time
performance_info = {
'execution_time': execution_time,
'rows_returned': len(result_df),
'query_plan': plan_df.to_dict('records')
}
print(f"执行时间: {execution_time:.4f}s")
print(f"返回行数: {len(result_df):,}")
return performance_info
except Exception as e:
print(f"❌ 性能分析失败: {e}")
return {}
def cleanup(self):
"""清理资源"""
if self.engine:
self.engine.dispose()
print("\n数据库连接已关闭")
# 实例化数据库管理器
db_manager = DatabaseManager()
# 创建示例数据
sample_data = db_manager.create_sample_data()
# 批量插入数据
for table_name, df in sample_data.items():
db_manager.bulk_insert_optimized(table_name, df)
# 创建索引
index_definitions = [
{'name': 'idx_users_status', 'table': 'users', 'columns': 'status'},
{'name': 'idx_orders_user_id', 'table': 'orders', 'columns': 'user_id'},
{'name': 'idx_orders_date', 'table': 'orders', 'columns': 'order_date'},
{'name': 'idx_products_category', 'table': 'products', 'columns': 'category'}
]
db_manager.create_indexes(index_definitions)
# 执行复杂查询
complex_query = """
SELECT
u.status,
COUNT(DISTINCT u.user_id) as user_count,
COUNT(o.order_id) as total_orders,
SUM(o.price * o.quantity) as total_revenue,
AVG(o.price * o.quantity) as avg_order_value,
MAX(o.order_date) as last_order_date
FROM users u
LEFT JOIN orders o ON u.user_id = o.user_id
WHERE o.status = 'completed'
GROUP BY u.status
ORDER BY total_revenue DESC
"""
result_df = db_manager.execute_complex_query(complex_query)
if not result_df.empty:
print("\n查询结果:")
print(result_df)
# 性能分析
performance_query = """
SELECT
p.category,
SUM(o.quantity) as total_sold,
SUM(o.price * o.quantity) as category_revenue
FROM products p
JOIN orders o ON p.product_id = o.product_id
WHERE o.status = 'completed' AND o.order_date >= '2024-06-01'
GROUP BY p.category
HAVING category_revenue > 1000
ORDER BY category_revenue DESC
"""
performance_info = db_manager.analyze_performance(performance_query)
# 清理资源
db_manager.cleanup()
# 删除数据库文件
db_file = 'advanced_demo.db'
if Path(db_file).exists():
Path(db_file).unlink()
print(f"已清理数据库文件: {db_file}")
并发文件访问与锁机制
文件锁定与多进程安全
import fcntl
import portalocker
import threading
from concurrent.futures import ThreadPoolExecutor, ProcessPoolExecutor
import multiprocessing
import queue
import tempfile
import shutil
print("=== 并发文件访问控制系统 ===")
class ConcurrentFileManager:
"""并发文件访问管理器
功能特性:
- 文件锁定机制
- 多进程文件访问示例，当前锁文件生命周期存在竞争风险
- 原地覆盖写入，不保证原子替换
- 尚未实现锁超时及死锁检测
- 共享读锁与独占写锁的使用示例，写操作不应使用共享锁
"""
def __init__(self, base_dir: str = None):
self.base_dir = Path(base_dir) if base_dir else Path(tempfile.mkdtemp())
self.lock_timeout = 30 # 锁超时时间（秒）
self._ensure_directory()
def _ensure_directory(self):
"""确保目录存在"""
self.base_dir.mkdir(parents=True, exist_ok=True)
print(f"工作目录: {self.base_dir}")
def write_with_lock(self, filename: str, data: str, exclusive: bool = True) -> bool:
"""使用文件锁的写操作"""
file_path = self.base_dir / filename
lock_path = self.base_dir / f"{filename}.lock"
try:
with open(lock_path, 'w') as lock_file:
# 获取独占锁
if exclusive:
portalocker.lock(lock_file, portalocker.LOCK_EX)
else:
portalocker.lock(lock_file, portalocker.LOCK_SH)
print(f"[{threading.current_thread().name}] 获取锁: {filename}")
# 原地覆盖写入；这不是临时文件加原子替换
with open(file_path, 'w') as f:
f.write(data)
f.flush() # 将Python缓冲区的数据提交给底层文件对象；后续调用fsync请求同步
os.fsync(f.fileno())
# 模拟一些处理时间
time.sleep(0.1)
print(f"[{threading.current_thread().name}] 写入完成: {filename}")
return True
except Exception as e:
print(f"❌ 写入失败 {filename}: {e}")
return False
finally:
# 锁会在with语句结束时自动释放
if lock_path.exists():
lock_path.unlink()
def read_with_lock(self, filename: str) -> Optional[str]:
"""使用文件锁的读操作"""
file_path = self.base_dir / filename
lock_path = self.base_dir / f"{filename}.lock"
if not file_path.exists():
return None
try:
with open(lock_path, 'w') as lock_file:
# 获取共享锁
portalocker.lock(lock_file, portalocker.LOCK_SH)
print(f"[{threading.current_thread().name}] 获取读锁: {filename}")
with open(file_path, 'r') as f:
content = f.read()
print(f"[{threading.current_thread().name}] 读取完成: {filename}")
return content
except Exception as e:
print(f"❌ 读取失败 {filename}: {e}")
return None
finally:
if lock_path.exists():
lock_path.unlink()
def demonstrate_concurrent_write(self, num_threads: int = 5):
"""演示并发写入"""
print(f"\n=== 并发写入测试 ({num_threads}个线程) ===")
def worker(thread_id):
filename = f"concurrent_test_{thread_id}.txt"
data = f"Thread {thread_id} data: {time.time()}\n" * 100
success = self.write_with_lock(filename, data)
return thread_id, success
with ThreadPoolExecutor(max_workers=num_threads) as executor:
futures = [executor.submit(worker, i) for i in range(num_threads)]
results = []
for future in futures:
thread_id, success = future.result()
results.append((thread_id, success))
print(f"线程 {thread_id}: {'成功' if success else '失败'}")
return results
def demonstrate_producer_consumer(self):
"""演示生产者-消费者模式的文件处理"""
print(f"\n=== 生产者-消费者文件处理 ===")
task_queue = queue.Queue(maxsize=10)
result_queue = queue.Queue()
def producer():
"""生产者：生成数据并写入文件"""
for i in range(20):
data = {
'id': i,
'timestamp': time.time(),
'data': f'数据内容 {i}' * 50
}
filename = f"producer_data_{i:03d}.json"
# 将数据序列化并写入文件
with open(self.base_dir / filename, 'w') as f:
import json
json.dump(data, f)
task_queue.put(filename)
print(f"生产者: 生成 {filename}")
time.sleep(0.05) # 模拟生产时间
# 添加结束标志
for _ in range(3): # 3个消费者
task_queue.put(None)
def consumer(consumer_id):
"""消费者：读取并处理文件"""
processed_count = 0
while True:
filename = task_queue.get()
if filename is None: # 结束标志
print(f"消费者 {consumer_id}: 处理结束，共处理 {processed_count} 个文件")
break
# 读取并处理文件
content = self.read_with_lock(filename)
if content:
# 模拟处理时间
time.sleep(0.02)
processed_count += 1
# 处理完成后删除文件
file_path = self.base_dir / filename
if file_path.exists():
file_path.unlink()
print(f"消费者 {consumer_id}: 处理完成 {filename}")
task_queue.task_done()
result_queue.put(('consumer', consumer_id, processed_count))
# 启动生产者和消费者线程
with ThreadPoolExecutor(max_workers=4) as executor:
# 1个生产者
producer_future = executor.submit(producer)
# 3个消费者
consumer_futures = [
executor.submit(consumer, i) for i in range(3)
]
# 等待所有任务完成
producer_future.result()
for future in consumer_futures:
future.result()
# 收集结果
results = []
while not result_queue.empty():
results.append(result_queue.get())
print(f"\n生产者-消费者处理完成:")
for result in results:
print(f" {result[0]} {result[1]}: 处理了 {result[2]} 个文件")
def cleanup(self):
"""清理工作目录"""
if self.base_dir.exists():
shutil.rmtree(self.base_dir)
print(f"\n已清理工作目录: {self.base_dir}")
# 演示并发文件访问控制
file_manager = ConcurrentFileManager()
# 演示并发写入
write_results = file_manager.demonstrate_concurrent_write(num_threads=8)
# 演示生产者-消费者模式
file_manager.demonstrate_producer_consumer()
# 清理资源
file_manager.cleanup()
性能优化与最佳实践
内存映射与零拷贝技术
import mmap
import struct
from contextlib import contextmanager
print("=== 高性能文件访问技术 ===")
class HighPerformanceFileProcessor:
"""高性能文件处理器
技术特性:
- 内存映射文件访问
- NumPy读取字节缓冲区；当前mmap切片仍会复制数据
- 异步I/O操作
- NumPy向量化字节统计，未独立实现或验证SIMD优化
- 内存映射访问示例，不涉及设备级直接内存访问
"""
def __init__(self):
self.chunk_size = 64 * 1024 * 1024 # 64MB chunks
self.buffer_size = 8192 # 8KB buffer
@contextmanager
def memory_mapped_file(self, filename: str, mode: str = 'r+b'):
"""内存映射文件上下文管理器"""
file_obj = None
mmap_obj = None
try:
file_obj = open(filename, mode)
mmap_obj = mmap.mmap(file_obj.fileno(), 0)
yield mmap_obj
finally:
if mmap_obj:
mmap_obj.close()
if file_obj:
file_obj.close()
def create_large_binary_file(self, filename: str, size_mb: int = 100):
"""创建大型二进制测试文件"""
print(f"创建 {size_mb}MB 测试文件: {filename}")
start_time = time.time()
with open(filename, 'wb') as f:
# 生成随机数据
data = np.random.randint(0, 256, size=1024*1024, dtype=np.uint8)
for i in range(size_mb):
f.write(data.tobytes())
if i % 10 == 0:
print(f" 已写入: {i+1}/{size_mb} MB")
creation_time = time.time() - start_time
print(f"✅ 文件创建完成，耗时: {creation_time:.3f}s")
def process_with_memory_map(self, filename: str):
"""使用内存映射处理大文件"""
print(f"\n使用内存映射处理文件: {filename}")
start_time = time.time()
try:
with self.memory_mapped_file(filename) as mmap_obj:
file_size = len(mmap_obj)
print(f"文件大小: {file_size:,} bytes ({file_size/1024/1024:.2f} MB)")
# 统计字节值分布
byte_counts = np.zeros(256, dtype=np.int64)
# 分块处理避免内存溢出
chunk_size = 1024 * 1024 # 1MB chunks
total_processed = 0
for i in range(0, file_size, chunk_size):
end_pos = min(i + chunk_size, file_size)
chunk = mmap_obj[i:end_pos]
# 将字节数据转换为numpy数组进行快速处理
np_chunk = np.frombuffer(chunk, dtype=np.uint8)
# 使用numpy的bincount进行高效统计
counts = np.bincount(np_chunk, minlength=256)
byte_counts += counts
total_processed += len(chunk)
if i % (10 * 1024 * 1024) == 0: # 每10MB报告一次
progress = (total_processed / file_size) * 100
print(f" 处理进度: {progress:.1f}% ({total_processed:,}/{file_size:,})")
processing_time = time.time() - start_time
# 分析结果
total_bytes = np.sum(byte_counts)
most_common = np.argmax(byte_counts)
present_values = np.flatnonzero(byte_counts > 0)
least_common = present_values[np.argmin(byte_counts[present_values])]
print(f"\n处理结果:")
print(f"- 处理时间: {processing_time:.3f}s")
print(f"- 处理速度: {total_bytes/processing_time/1024/1024:.2f} MB/s")
print(f"- 最常见字节值: {most_common} (出现 {byte_counts[most_common]:,} 次)")
print(f"- 最少见字节值: {least_common} (出现 {byte_counts[least_common]:,} 次)")
print(f"- 字节值分布熵: {self._calculate_entropy(byte_counts):.4f}")
except Exception as e:
print(f"❌ 内存映射处理失败: {e}")
def process_with_regular_io(self, filename: str):
"""使用常规I/O处理文件进行对比"""
print(f"\n使用常规I/O处理文件: {filename}")
start_time = time.time()
try:
file_size = Path(filename).stat().st_size
byte_counts = np.zeros(256, dtype=np.int64)
total_processed = 0
with open(filename, 'rb') as f:
while True:
chunk = f.read(1024 * 1024) # 1MB chunks
if not chunk:
break
# 转换为numpy数组处理
np_chunk = np.frombuffer(chunk, dtype=np.uint8)
counts = np.bincount(np_chunk, minlength=256)
byte_counts += counts
total_processed += len(chunk)
if total_processed % (10 * 1024 * 1024) == 0:
progress = (total_processed / file_size) * 100
print(f" 处理进度: {progress:.1f}%")
processing_time = time.time() - start_time
print(f"常规I/O处理结果:")
print(f"- 处理时间: {processing_time:.3f}s")
print(f"- 处理速度: {total_processed/processing_time/1024/1024:.2f} MB/s")
except Exception as e:
print(f"❌ 常规I/O处理失败: {e}")
def demonstrate_async_io(self, filenames: List[str]):
"""演示异步I/O操作"""
print(f"\n=== 异步I/O处理演示 ===")
async def process_file_async(filename: str) -> Dict[str, Any]:
"""异步处理单个文件"""
import aiofiles
start_time = time.time()
file_size = 0
line_count = 0
try:
async with aiofiles.open(filename, 'rb') as f:
while True:
chunk = await f.read(8192)
if not chunk:
break
file_size += len(chunk)
line_count += chunk.count(b'\n')
processing_time = time.time() - start_time
return {
'filename': filename,
'size': file_size,
'lines': line_count,
'time': processing_time,
'speed_mbps': file_size / processing_time / 1024 / 1024
}
except Exception as e:
return {
'filename': filename,
'error': str(e)
}
async def process_all_files():
"""并发处理所有文件"""
tasks = [process_file_async(filename) for filename in filenames]
results = await asyncio.gather(*tasks, return_exceptions=True)
return results
# 运行异步处理
try:
results = asyncio.run(process_all_files())
print(f"异步I/O处理完成:")
total_time = 0
total_size = 0
for result in results:
if isinstance(result, dict) and 'error' not in result:
print(f"- {result['filename']}: "
f"{result['size']:,} bytes, "
f"{result['lines']:,} lines, "
f"{result['time']:.3f}s, "
f"{result['speed_mbps']:.2f} MB/s")
total_time = max(total_time, result['time']) # 并发执行，取最长时间
total_size += result['size']
else:
print(f"- 错误: {result}")
if total_time > 0:
overall_speed = total_size / total_time / 1024 / 1024
print(f"\n总体性能: {overall_speed:.2f} MB/s (并发处理)")
except ImportError:
print("❌ aiofiles模块未安装，跳过异步I/O演示")
except Exception as e:
print(f"❌ 异步I/O处理失败: {e}")
def _calculate_entropy(self, counts: np.ndarray) -> float:
"""计算数据分布的熵"""
total = np.sum(counts)
if total == 0:
return 0.0
probabilities = counts[counts > 0] / total
entropy = -np.sum(probabilities * np.log2(probabilities))
return entropy
def cleanup_test_files(self, filenames: List[str]):
"""清理测试文件"""
print(f"\n清理测试文件...")
for filename in filenames:
if Path(filename).exists():
Path(filename).unlink()
print(f"已删除: {filename}")
# 演示高性能文件处理
processor = HighPerformanceFileProcessor()
# 创建测试文件
test_file = 'large_binary_test.dat'
processor.create_large_binary_file(test_file, size_mb=50)
# 性能对比测试
print("\n=== 性能对比测试 ===")
# 内存映射处理
processor.process_with_memory_map(test_file)
# 常规I/O处理
processor.process_with_regular_io(test_file)
# 创建多个小文件进行异步I/O测试
small_files = []
for i in range(5):
filename = f'async_test_{i}.txt'
with open(filename, 'w') as f:
f.write("test data line\n" * 10000)
small_files.append(filename)
# 异步I/O演示
processor.demonstrate_async_io(small_files)
# 清理所有测试文件
all_test_files = [test_file] + small_files
processor.cleanup_test_files(all_test_files)
数据工程最佳实践
ETL管道与数据质量保证
from dataclasses import dataclass
from enum import Enum
import hashlib
from typing import Protocol, runtime_checkable
print("=== 数据工程ETL管道系统 ===")
class DataQuality(Enum):
"""数据质量等级"""
EXCELLENT = "excellent"
GOOD = "good"
ACCEPTABLE = "acceptable"
POOR = "poor"
UNACCEPTABLE = "unacceptable"
@dataclass
class DataQualityReport:
"""数据质量报告"""
total_records: int
valid_records: int
missing_values: int
duplicate_records: int
outliers: int
quality_score: float
quality_level: DataQuality
issues: List[str]
recommendations: List[str]
@runtime_checkable
class DataProcessor(Protocol):
"""数据处理器协议"""
def process(self, data: pd.DataFrame) -> pd.DataFrame:
...
def validate(self, data: pd.DataFrame) -> bool:
...
class ETLPipeline:
"""ETL数据管道教学示例
功能特性:
- 数据提取、转换、加载
- 数据质量监控
- 错误捕获，尚未实现自动重试
- 阶段级数据血缘记录
- 执行时间统计
- 尚未实现增量更新
"""
def __init__(self, pipeline_name: str):
self.pipeline_name = pipeline_name
self.processors: List[DataProcessor] = []
self.quality_checks: List[callable] = []
self.metadata = {
'created_at': time.time(),
'runs': 0,
'last_run': None,
'success_rate': 0.0
}
self.data_lineage = []
def add_processor(self, processor: DataProcessor):
"""添加数据处理器"""
self.processors.append(processor)
print(f"添加处理器: {processor.__class__.__name__}")
def add_quality_check(self, check_func: callable):
"""添加数据质量检查"""
self.quality_checks.append(check_func)
print(f"添加质量检查: {check_func.__name__}")
def extract_data(self, sources: List[Dict[str, Any]]) -> pd.DataFrame:
"""数据提取阶段"""
print(f"\n=== 数据提取阶段 ===")
all_data = []
for source in sources:
source_type = source['type']
source_path = source['path']
print(f"提取数据源: {source_type} - {source_path}")
try:
if source_type == 'csv':
df = pd.read_csv(source_path)
elif source_type == 'excel':
df = pd.read_excel(source_path)
elif source_type == 'json':
df = pd.read_json(source_path)
elif source_type == 'database':
# 这里应该连接数据库
df = pd.DataFrame() # 占位
else:
raise ValueError(f"不支持的数据源类型: {source_type}")
# 添加元数据
df['_source'] = source_path
df['_extracted_at'] = pd.Timestamp.now()
all_data.append(df)
print(f" 成功提取 {len(df)} 行数据")
# 记录数据血缘
self.data_lineage.append({
'stage': 'extract',
'source': source_path,
'rows': len(df),
'timestamp': time.time()
})
except Exception as e:
print(f" ❌ 提取失败: {e}")
raise
if all_data:
combined_df = pd.concat(all_data, ignore_index=True)
print(f"数据提取完成: 总计 {len(combined_df)} 行")
return combined_df
else:
return pd.DataFrame()
def transform_data(self, data: pd.DataFrame) -> pd.DataFrame:
"""数据转换阶段"""
print(f"\n=== 数据转换阶段 ===")
if data.empty:
return data
current_data = data.copy()
for i, processor in enumerate(self.processors):
print(f"执行处理器 {i+1}/{len(self.processors)}: {processor.__class__.__name__}")
try:
# 验证输入数据
if hasattr(processor, 'validate') and not processor.validate(current_data):
raise ValueError("数据验证失败")
# 处理数据
start_time = time.time()
processed_data = processor.process(current_data)
processing_time = time.time() - start_time
print(f" 处理完成: {len(current_data)} -> {len(processed_data)} 行, "
f"耗时 {processing_time:.3f}s")
current_data = processed_data
# 记录数据血缘
self.data_lineage.append({
'stage': 'transform',
'processor': processor.__class__.__name__,
'input_rows': len(data) if i == 0 else len(current_data),
'output_rows': len(processed_data),
'processing_time': processing_time,
'timestamp': time.time()
})
except Exception as e:
print(f" ❌ 处理失败: {e}")
raise
print(f"数据转换完成: {len(current_data)} 行")
return current_data
def validate_data_quality(self, data: pd.DataFrame) -> DataQualityReport:
"""数据质量验证"""
print(f"\n=== 数据质量验证 ===")
if data.empty:
return DataQualityReport(
total_records=0,
valid_records=0,
missing_values=0,
duplicate_records=0,
outliers=0,
quality_score=0.0,
quality_level=DataQuality.UNACCEPTABLE,
issues=["数据为空"],
recommendations=["检查数据源"]
)
# 基础质量指标
total_records = len(data)
missing_values = data.isnull().sum().sum()
duplicate_records = data.duplicated().sum()
# 检测异常值
numeric_cols = data.select_dtypes(include=[np.number]).columns
outliers = 0
for col in numeric_cols:
Q1 = data[col].quantile(0.25)
Q3 = data[col].quantile(0.75)
IQR = Q3 - Q1
outliers += ((data[col] < (Q1 - 1.5 * IQR)) |
(data[col] > (Q3 + 1.5 * IQR))).sum()
# 执行自定义质量检查
custom_issues = []
for check_func in self.quality_checks:
try:
result = check_func(data)
if not result['passed']:
custom_issues.append(result['message'])
except Exception as e:
custom_issues.append(f"质量检查失败: {e}")
# 计算质量分数
missing_ratio = missing_values / (total_records * len(data.columns))
duplicate_ratio = duplicate_records / total_records
outlier_ratio = outliers / (total_records * len(numeric_cols)) if len(numeric_cols) > 0 else 0
quality_score = max(0.0, 1.0 - (missing_ratio * 0.4 + duplicate_ratio * 0.3 + outlier_ratio * 0.3))
# 确定质量等级
if quality_score >= 0.9:
quality_level = DataQuality.EXCELLENT
elif quality_score >= 0.8:
quality_level = DataQuality.GOOD
elif quality_score >= 0.7:
quality_level = DataQuality.ACCEPTABLE
elif quality_score >= 0.5:
quality_level = DataQuality.POOR
else:
quality_level = DataQuality.UNACCEPTABLE
valid_records = total_records - duplicate_records
# 生成问题列表
issues = custom_issues.copy()
if missing_ratio > 0.1:
issues.append(f"缺失值比例过高: {missing_ratio:.2%}")
if duplicate_ratio > 0.05:
issues.append(f"重复记录比例过高: {duplicate_ratio:.2%}")
if outlier_ratio > 0.1:
issues.append(f"异常值比例过高: {outlier_ratio:.2%}")
# 生成建议
recommendations = []
if missing_values > 0:
recommendations.append("处理缺失值：填充、删除或插值")
if duplicate_records > 0:
recommendations.append("删除重复记录")
if outliers > 0:
recommendations.append("检查并处理异常值")
if quality_score < 0.7:
recommendations.append("需要提升数据源质量")
report = DataQualityReport(
total_records=total_records,
valid_records=valid_records,
missing_values=missing_values,
duplicate_records=duplicate_records,
outliers=outliers,
quality_score=quality_score,
quality_level=quality_level,
issues=issues,
recommendations=recommendations
)
print(f"数据质量报告:")
print(f"- 总记录数: {total_records:,}")
print(f"- 有效记录: {valid_records:,}")
print(f"- 缺失值: {missing_values:,}")
print(f"- 重复记录: {duplicate_records:,}")
print(f"- 异常值: {outliers:,}")
print(f"- 质量分数: {quality_score:.3f}")
print(f"- 质量等级: {quality_level.value}")
if issues:
print("- 发现问题:")
for issue in issues:
print(f" • {issue}")
return report
def load_data(self, data: pd.DataFrame, target: Dict[str, Any]) -> bool:
"""数据加载阶段"""
print(f"\n=== 数据加载阶段 ===")
if data.empty:
print("❌ 数据为空，跳过加载")
return False
target_type = target['type']
target_path = target['path']
try:
print(f"加载到目标: {target_type} - {target_path}")
if target_type == 'csv':
data.to_csv(target_path, index=False)
elif target_type == 'excel':
data.to_excel(target_path, index=False)
elif target_type == 'parquet':
data.to_parquet(target_path, index=False)
elif target_type == 'database':
raise NotImplementedError("当前示例尚未实现数据库加载")
else:
raise ValueError(f"不支持的目标类型: {target_type}")
print(f"✅ 成功加载 {len(data)} 行数据到 {target_path}")
# 记录数据血缘
self.data_lineage.append({
'stage': 'load',
'target': target_path,
'rows': len(data),
'timestamp': time.time()
})
return True
except Exception as e:
print(f"❌ 加载失败: {e}")
return False
def run_pipeline(self, sources: List[Dict[str, Any]],
target: Dict[str, Any]) -> Dict[str, Any]:
"""运行完整ETL管道"""
print(f"\n========== 运行ETL管道: {self.pipeline_name} ==========")
pipeline_start = time.time()
try:
# 更新运行统计
self.metadata['runs'] += 1
self.metadata['last_run'] = pipeline_start
# 1. 数据提取
raw_data = self.extract_data(sources)
# 2. 数据转换
transformed_data = self.transform_data(raw_data)
# 3. 数据质量验证
quality_report = self.validate_data_quality(transformed_data)
# 检查质量是否合格
if quality_report.quality_level in [DataQuality.UNACCEPTABLE, DataQuality.POOR]:
print(f"❌ 数据质量不合格: {quality_report.quality_level.value}")
return {
'success': False,
'error': '数据质量不符合要求',
'quality_report': quality_report
}
# 4. 数据加载
load_success = self.load_data(transformed_data, target)
pipeline_time = time.time() - pipeline_start
if load_success:
print(f"\n✅ ETL管道执行成功")
print(f"执行时间: {pipeline_time:.3f}s")
print(f"处理记录: {len(transformed_data):,}")
print(f"处理速度: {len(transformed_data)/pipeline_time:,.0f} 记录/秒")
# 更新成功率
success_count = getattr(self, '_success_count', 0) + 1
self._success_count = success_count
self.metadata['success_rate'] = success_count / self.metadata['runs']
return {
'success': True,
'execution_time': pipeline_time,
'records_processed': len(transformed_data),
'quality_report': quality_report,
'data_lineage': self.data_lineage
}
else:
raise Exception("数据加载失败")
except Exception as e:
pipeline_time = time.time() - pipeline_start
print(f"\n❌ ETL管道执行失败: {e}")
print(f"执行时间: {pipeline_time:.3f}s")
return {
'success': False,
'error': str(e),
'execution_time': pipeline_time
}
def get_pipeline_stats(self) -> Dict[str, Any]:
"""获取管道统计信息"""
return {
'pipeline_name': self.pipeline_name,
'metadata': self.metadata,
'processors_count': len(self.processors),
'quality_checks_count': len(self.quality_checks),
'data_lineage_entries': len(self.data_lineage)
}
# 定义示例数据处理器
class DataCleaningProcessor:
"""数据清理处理器"""
def validate(self, data: pd.DataFrame) -> bool:
return not data.empty
def process(self, data: pd.DataFrame) -> pd.DataFrame:
# 删除完全空的行
cleaned = data.dropna(how='all')
# 删除重复行
cleaned = cleaned.drop_duplicates()
# 基本数据类型推断
for col in cleaned.select_dtypes(include=['object']):
# 尝试转换为数值类型
numeric_converted = pd.to_numeric(cleaned[col], errors='coerce')
if numeric_converted.notna().sum() > len(cleaned) * 0.8:
cleaned[col] = numeric_converted
return cleaned
class DataEnrichmentProcessor:
"""数据增强处理器"""
def validate(self, data: pd.DataFrame) -> bool:
return 'timestamp' in data.columns or '_extracted_at' in data.columns
def process(self, data: pd.DataFrame) -> pd.DataFrame:
enriched = data.copy()
# 添加处理时间戳
enriched['processed_at'] = pd.Timestamp.now()
# 添加数据hash用于去重检查
enriched['data_hash'] = enriched.apply(
lambda row: hashlib.md5(str(row.values).encode()).hexdigest(),
axis=1
)
# 添加质量标记
enriched['quality_flag'] = 'processed'
return enriched
# 定义质量检查函数
def check_completeness(data: pd.DataFrame) -> Dict[str, Any]:
"""检查数据完整性"""
missing_ratio = data.isnull().sum().sum() / (len(data) * len(data.columns))
return {
'passed': missing_ratio < 0.1,
'message': f'数据完整性检查: 缺失率 {missing_ratio:.2%}'
}
def check_consistency(data: pd.DataFrame) -> Dict[str, Any]:
"""检查数据一致性"""
# 简单的一致性检查：数值列不应有极端异常值
numeric_cols = data.select_dtypes(include=[np.number]).columns
inconsistent_count = 0
for col in numeric_cols:
if col in data.columns:
std = data[col].std()
mean = data[col].mean()
if std > 0:
z_scores = np.abs((data[col] - mean) / std)
inconsistent_count += (z_scores > 5).sum() # 5个标准差外
inconsistency_ratio = inconsistent_count / len(data) if len(data) > 0 else 0
return {
'passed': inconsistency_ratio < 0.05,
'message': f'数据一致性检查: 异常值比例 {inconsistency_ratio:.2%}'
}
# 演示ETL管道
print("=== ETL管道演示 ===")
# 创建测试数据
test_data = pd.DataFrame({
'id': range(1000),
'name': [f'用户{i}' for i in range(1000)],
'age': np.random.randint(18, 80, 1000),
'salary': np.random.normal(50000, 15000, 1000),
'department': np.random.choice(['技术', '销售', '市场', '人事'], 1000),
'join_date': pd.date_range('2020-01-01', periods=1000, freq='D')
})
# 引入一些数据质量问题
test_data.loc[50:60, 'salary'] = np.nan # 缺失值
test_data = pd.concat([test_data, test_data.head(10)]) # 重复值
test_data.loc[100:105, 'salary'] = test_data.loc[100:105, 'salary'] * 10 # 异常值
# 保存测试数据
test_csv = 'etl_test_data.csv'
test_data.to_csv(test_csv, index=False)
# 创建ETL管道
pipeline = ETLPipeline("用户数据处理管道")
# 添加处理器
pipeline.add_processor(DataCleaningProcessor())
pipeline.add_processor(DataEnrichmentProcessor())
# 添加质量检查
pipeline.add_quality_check(check_completeness)
pipeline.add_quality_check(check_consistency)
# 配置数据源和目标
sources = [
{
'type': 'csv',
'path': test_csv
}
]
target = {
'type': 'parquet',
'path': 'etl_output.parquet'
}
# 运行ETL管道
result = pipeline.run_pipeline(sources, target)
# 显示结果
print(f"\n=== ETL执行结果 ===")
print(f"执行结果: {'成功' if result['success'] else '失败'}")
if result['success']:
print(f"处理记录数: {result['records_processed']:,}")
print(f"执行时间: {result['execution_time']:.3f}s")
print(f"数据质量: {result['quality_report'].quality_level.value}")
else:
print(f"失败原因: {result['error']}")
# 显示管道统计
stats = pipeline.get_pipeline_stats()
print(f"\n=== 管道统计信息 ===")
print(f"管道名称: {stats['pipeline_name']}")
print(f"运行次数: {stats['metadata']['runs']}")
print(f"成功率: {stats['metadata']['success_rate']:.2%}")
print(f"处理器数量: {stats['processors_count']}")
print(f"质量检查数量: {stats['quality_checks_count']}")
# 清理测试文件
test_files = [test_csv, 'etl_output.parquet']
for file in test_files:
if Path(file).exists():
Path(file).unlink()
print(f"已清理文件: {file}")
总结与最佳实践
文件I/O性能优化核心要点
-
缓冲策略选择
- 大文件使用大缓冲区(64KB-1MB)
- 频繁小写操作使用小缓冲区(4KB-8KB)
- 顺序访问优化缓冲区大小
- 随机访问考虑内存映射
-
数据格式选择原则
- CSV: 通用性好，调试友好，但性能一般
- Parquet: 列式存储，压缩率高，查询性能优秀
- Feather: 快速读写，适合临时存储
- Excel: 商务友好，支持格式化，但性能较低
-
大文件处理策略
- 流式处理避免内存溢出
- 分块并行处理提升效率
- 内存映射减少数据拷贝
- 异步I/O提高并发性能
-
数据库连接优化
- 连接池管理避免频繁连接
- 批量操作减少网络开销
- 预编译语句提高安全性
- 读写分离优化查询性能
-
并发安全控制
- 文件锁机制保证数据一致性
- 原子性操作避免数据损坏
- 适当的锁粒度平衡性能和安全
- 死锁检测和预防机制
通过掌握这些核心技术和最佳实践，你将能够构建高效、可靠的数据处理系统，为大规模数据工程项目提供坚实的技术基础。
学习资源连接：
- 基础学习: Python基础语法 → Python数据结构
- 进阶应用: Pandas数据操作 → Pandas数据清洗
- 实战项目: 端到端项目 → 综合项目
- 相关技术: Hadoop集群规划
标签: #Python #文件IO #数据存储 #性能优化 #数据工程 #ETL管道 #数据质量