Python NumPy批量处理CSV文件数据的最佳实践教程
作者：知远漫谈
在数据分析和科学计算的世界中,CSV文件是最常见的数据存储格式之一,本文将深入探讨如何使用NumPy进行CSV文件的批量处理,从基础操作到高级技巧,帮助你提升数据处理效率,
在数据分析和科学计算的世界中，CSV文件是最常见的数据存储格式之一。无论是金融数据、销售记录还是实验结果，我们经常需要处理大量的CSV文件。Python的NumPy库作为数值计算的核心工具，配合pandas等库，能够高效地处理这些批量数据任务。本文将深入探讨如何使用NumPy进行CSV文件的批量处理，从基础操作到高级技巧，帮助你提升数据处理效率。
为什么选择NumPy进行CSV处理？
NumPy是Python科学计算的基础库，它提供了高性能的多维数组对象和相关工具。虽然pandas在数据处理方面更加直观，但NumPy在以下场景下具有独特优势：
- 内存效率：NumPy数组比Python列表更节省内存
- 计算速度：向量化操作比循环快得多
- 数学函数：丰富的数学运算函数
- 与其他库兼容性好：SciPy、matplotlib等都基于NumPy
让我们开始实际的编码之旅吧
基础环境准备
首先，确保安装了必要的库：
import numpy as np import pandas as pd import os import glob from pathlib import Path import time
理解CSV数据结构
在处理CSV文件之前，我们需要理解其基本结构。CSV文件本质上是以逗号分隔的纯文本文件，每一行代表一条记录，每个字段由逗号分隔。
让我们创建一些示例数据来演示：
# 创建示例CSV文件
def create_sample_csv_files():
"""创建示例CSV文件用于演示"""
# 示例数据1：销售数据
sales_data = [
['日期', '产品', '销售额', '数量'],
['2023-01-01', '产品A', 1000, 50],
['2023-01-01', '产品B', 1500, 75],
['2023-01-02', '产品A', 1200, 60],
['2023-01-02', '产品C', 800, 40]
]
# 示例数据2：温度数据
temp_data = [
['时间', '温度', '湿度'],
['08:00', 22.5, 65],
['12:00', 28.3, 55],
['16:00', 26.8, 60],
['20:00', 24.1, 70]
]
# 写入文件
with open('sales_2023.csv', 'w', encoding='utf-8') as f:
for row in sales_data:
f.write(','.join(map(str, row)) + '\n')
with open('temp_2023.csv', 'w', encoding='utf-8') as f:
for row in temp_data:
f.write(','.join(map(str, row)) + '\n')
create_sample_csv_files()
使用NumPy读取单个CSV文件
NumPy提供了loadtxt和genfromtxt两个主要函数来读取CSV文件：
# 方法1：使用loadtxt（适用于简单情况）
try:
# 跳过标题行，指定分隔符
data = np.loadtxt('sales_2023.csv', delimiter=',', skiprows=1,
dtype={'names': ('date', 'product', 'sales', 'quantity'),
'formats': ('U10', 'U10', 'f4', 'i4')})
print("使用loadtxt读取的数据:")
print(data)
except Exception as e:
print(f"loadtxt方法遇到问题: {e}")
# 方法2：使用genfromtxt（更灵活）
data_array = np.genfromtxt('sales_2023.csv', delimiter=',', skip_header=1,
filling_values=0, dtype=None, encoding='utf-8')
print("\n使用genfromtxt读取的数据:")
print(data_array)
# 方法3：混合使用pandas和NumPy（推荐）
df = pd.read_csv('sales_2023.csv')
numpy_array = df.values # 转换为NumPy数组
print("\n使用pandas+NumPy读取的数据:")
print(numpy_array)
批量处理多个CSV文件
在实际工作中，我们经常需要处理同一目录下的多个CSV文件。以下是几种常用的批量处理方法：
方法一：使用glob模块
def process_multiple_csv_with_glob(pattern="*.csv"):
"""使用glob模块批量处理CSV文件"""
# 获取所有匹配的CSV文件
csv_files = glob.glob(pattern)
print(f"找到 {len(csv_files)} 个CSV文件")
all_data = []
for file_path in csv_files:
try:
# 读取数据
data = np.genfromtxt(file_path, delimiter=',', skip_header=1,
filling_values=0, dtype=float)
# 处理数据（示例：计算每列的平均值）
if data.size > 0:
means = np.mean(data, axis=0) if len(data.shape) > 1 else data
all_data.append({
'file': file_path,
'means': means,
'shape': data.shape
})
print(f"✅ 处理完成: {file_path}")
else:
print(f"⚠️ 空文件: {file_path}")
except Exception as e:
print(f"❌ 处理失败 {file_path}: {e}")
return all_data
# 执行批量处理
results = process_multiple_csv_with_glob()
for result in results:
print(f"文件: {result['file']}, 形状: {result['shape']}")
方法二：使用pathlib模块
def process_multiple_csv_with_pathlib(directory="."):
"""使用pathlib模块批量处理CSV文件"""
path = Path(directory)
csv_files = list(path.glob("*.csv"))
print(f"在目录 '{directory}' 中找到 {len(csv_files)} 个CSV文件")
processed_data = {}
for csv_file in csv_files:
try:
# 读取CSV文件
df = pd.read_csv(csv_file)
numpy_data = df.select_dtypes(include=[np.number]).values
# 执行NumPy计算
stats = {
'mean': np.mean(numpy_data, axis=0) if numpy_data.size > 0 else np.array([]),
'std': np.std(numpy_data, axis=0) if numpy_data.size > 0 else np.array([]),
'min': np.min(numpy_data, axis=0) if numpy_data.size > 0 else np.array([]),
'max': np.max(numpy_data, axis=0) if numpy_data.size > 0 else np.array([])
}
processed_data[csv_file.name] = {
'stats': stats,
'original_shape': df.shape,
'numeric_shape': numpy_data.shape
}
print(f"✅ 成功处理: {csv_file.name}")
except Exception as e:
print(f"❌ 处理失败 {csv_file.name}: {e}")
return processed_data
# 执行处理
batch_results = process_multiple_csv_with_pathlib()
# 显示结果
for filename, data in batch_results.items():
print(f"\n📊 {filename}:")
print(f" 原始形状: {data['original_shape']}")
print(f" 数值数据形状: {data['numeric_shape']}")
if data['stats']['mean'].size > 0:
print(f" 平均值: {data['stats']['mean']}")
高级数据处理技术
数据清洗和预处理
def advanced_csv_processing(file_path):
"""高级CSV处理：包含数据清洗和预处理"""
try:
# 读取数据
df = pd.read_csv(file_path)
# 转换为NumPy数组进行数值计算
numeric_columns = df.select_dtypes(include=[np.number])
numeric_data = numeric_columns.values
print(f"原始数据形状: {df.shape}")
print(f"数值数据形状: {numeric_data.shape}")
# 数据质量检查
missing_values = df.isnull().sum()
print(f"缺失值统计:\n{missing_values}")
# 使用NumPy进行数据填充
if numeric_data.size > 0:
# 计算每列的中位数
medians = np.nanmedian(numeric_data, axis=0)
# 填充缺失值
filled_data = np.where(np.isnan(numeric_data),
medians,
numeric_data)
# 标准化数据
means = np.mean(filled_data, axis=0)
stds = np.std(filled_data, axis=0)
# 避免除零错误
stds = np.where(stds == 0, 1, stds)
normalized_data = (filled_data - means) / stds
return {
'original_data': numeric_data,
'filled_data': filled_data,
'normalized_data': normalized_data,
'medians': medians,
'means': means,
'stds': stds
}
else:
return None
except Exception as e:
print(f"处理失败: {e}")
return None
# 测试高级处理
result = advanced_csv_processing('sales_2023.csv')
if result:
print("数据处理完成!")
print(f"标准化后的数据形状: {result['normalized_data'].shape}")
性能优化技巧
当处理大量CSV文件时，性能优化变得至关重要：
def optimized_batch_processing(file_pattern="*.csv", chunk_size=1000):
"""优化的批量处理函数"""
start_time = time.time()
files = glob.glob(file_pattern)
total_rows_processed = 0
for file_path in files:
try:
# 分块读取大文件
chunk_reader = pd.read_csv(file_path, chunksize=chunk_size)
chunk_count = 0
for chunk in chunk_reader:
# 转换为NumPy数组
numeric_chunk = chunk.select_dtypes(include=[np.number]).values
if numeric_chunk.size > 0:
# 执行向量化计算
chunk_sum = np.sum(numeric_chunk, axis=0)
chunk_mean = np.mean(numeric_chunk, axis=0)
total_rows_processed += numeric_chunk.shape[0]
chunk_count += 1
if chunk_count % 10 == 0:
print(f"📦 处理进度: {file_path} - 第 {chunk_count} 块")
except Exception as e:
print(f"❌ 处理失败 {file_path}: {e}")
end_time = time.time()
processing_time = end_time - start_time
print(f"\n📈 处理统计:")
print(f" 总处理时间: {processing_time:.2f} 秒")
print(f" 总处理行数: {total_rows_processed:,}")
print(f" 处理速度: {total_rows_processed/processing_time:.0f} 行/秒")
# 执行优化处理
optimized_batch_processing()
数据聚合和分析
跨文件数据聚合
def aggregate_multiple_csv_files(file_pattern="*.csv"):
"""跨多个CSV文件进行数据聚合"""
files = glob.glob(file_pattern)
all_numeric_data = []
file_info = []
for file_path in files:
try:
df = pd.read_csv(file_path)
numeric_df = df.select_dtypes(include=[np.number])
if not numeric_df.empty:
numeric_data = numeric_df.values
all_numeric_data.append(numeric_data)
file_info.append({
'name': file_path,
'rows': numeric_data.shape[0],
'columns': numeric_data.shape[1],
'sum': np.sum(numeric_data),
'mean': np.mean(numeric_data)
})
except Exception as e:
print(f"处理文件失败 {file_path}: {e}")
if all_numeric_data:
# 合并所有数据
combined_data = np.vstack(all_numeric_data)
# 计算总体统计信息
overall_stats = {
'total_rows': combined_data.shape[0],
'total_columns': combined_data.shape[1],
'overall_mean': np.mean(combined_data),
'overall_std': np.std(combined_data),
'overall_min': np.min(combined_data),
'overall_max': np.max(combined_data),
'column_means': np.mean(combined_data, axis=0),
'column_stds': np.std(combined_data, axis=0)
}
return {
'combined_data': combined_data,
'file_info': file_info,
'overall_stats': overall_stats
}
return None
# 执行聚合分析
aggregation_result = aggregate_multiple_csv_files()
if aggregation_result:
stats = aggregation_result['overall_stats']
print("📊 聚合统计结果:")
print(f" 总行数: {stats['total_rows']:,}")
print(f" 总列数: {stats['total_columns']}")
print(f" 整体平均值: {stats['overall_mean']:.2f}")
print(f" 整体标准差: {stats['overall_std']:.2f}")
print(f" 最小值: {stats['overall_min']}")
print(f" 最大值: {stats['overall_max']}")
时间序列数据分析
def analyze_time_series_csv(file_path):
"""分析时间序列CSV数据"""
try:
df = pd.read_csv(file_path)
# 假设第一列是时间戳
if len(df.columns) > 1:
numeric_columns = df.select_dtypes(include=[np.number])
numeric_data = numeric_columns.values
if numeric_data.size > 0:
# 计算移动平均
def moving_average(data, window_size=3):
"""计算移动平均"""
if len(data) < window_size:
return data
weights = np.ones(window_size) / window_size
return np.convolve(data, weights, mode='valid')
# 对每列计算移动平均
ma_results = {}
for i, col_name in enumerate(numeric_columns.columns):
column_data = numeric_data[:, i]
ma_results[col_name] = moving_average(column_data)
# 计算增长率
growth_rates = {}
for i, col_name in enumerate(numeric_columns.columns):
column_data = numeric_data[:, i]
if len(column_data) > 1:
growth_rates[col_name] = np.diff(column_data) / column_data[:-1] * 100
return {
'original_data': numeric_data,
'moving_averages': ma_results,
'growth_rates': growth_rates,
'trend_analysis': np.polyfit(range(len(numeric_data)),
np.mean(numeric_data, axis=1), 1)
}
except Exception as e:
print(f"时间序列分析失败: {e}")
return None
# 测试时间序列分析
ts_result = analyze_time_series_csv('temp_2023.csv')
if ts_result:
print("⏰ 时间序列分析完成:")
for col, ma in ts_result['moving_averages'].items():
print(f" {col} 的移动平均: {ma}")
错误处理和异常管理
在批量处理CSV文件时，错误处理非常重要：
def robust_csv_processor(file_pattern="*.csv", max_errors=5):
"""健壮的CSV处理器，包含完善的错误处理"""
files = glob.glob(file_pattern)
error_count = 0
success_count = 0
results = {}
for file_path in files:
if error_count >= max_errors:
print("⚠️ 达到最大错误数，停止处理")
break
try:
# 尝试多种读取方式
df = None
# 方式1：标准读取
try:
df = pd.read_csv(file_path, encoding='utf-8')
except UnicodeDecodeError:
# 方式2：尝试其他编码
try:
df = pd.read_csv(file_path, encoding='gbk')
except:
# 方式3：忽略编码错误
df = pd.read_csv(file_path, encoding='utf-8', errors='ignore')
if df is not None and not df.empty:
# 转换为NumPy数组
numeric_data = df.select_dtypes(include=[np.number]).values
# 执行基本验证
if numeric_data.size > 0:
# 检查数据范围合理性
data_min = np.min(numeric_data)
data_max = np.max(numeric_data)
if np.isfinite(data_min) and np.isfinite(data_max):
results[file_path] = {
'status': 'success',
'rows': df.shape[0],
'numeric_rows': numeric_data.shape[0],
'numeric_cols': numeric_data.shape[1],
'data_range': (data_min, data_max),
'memory_usage': numeric_data.nbytes
}
success_count += 1
print(f"✅ 成功处理: {file_path}")
else:
raise ValueError("数据包含无效数值")
else:
raise ValueError("没有数值数据")
else:
raise ValueError("无法读取有效数据")
except Exception as e:
error_count += 1
results[file_path] = {
'status': 'error',
'error_message': str(e)
}
print(f"❌ 处理失败 {file_path}: {e}")
# 输出处理报告
print(f"\n📋 处理报告:")
print(f" 成功文件数: {success_count}")
print(f" 失败文件数: {error_count}")
print(f" 总文件数: {len(files)}")
return results
# 执行健壮处理
robust_results = robust_csv_processor()
# 显示成功处理的文件信息
successful_files = {k: v for k, v in robust_results.items() if v['status'] == 'success'}
for file_path, info in successful_files.items():
print(f"📄 {file_path}: {info['rows']} 行, {info['numeric_cols']} 列")
数据导出和保存
处理完数据后，通常需要将结果保存回文件：
def save_processed_data(processed_results, output_dir="processed_output"):
"""保存处理后的数据"""
# 创建输出目录
os.makedirs(output_dir, exist_ok=True)
for file_path, data in processed_results.items():
if data['status'] == 'success':
try:
# 构造输出文件名
base_name = os.path.splitext(os.path.basename(file_path))[0]
output_file = os.path.join(output_dir, f"{base_name}_processed.csv")
# 如果有数值数据，保存统计信息
original_df = pd.read_csv(file_path)
numeric_df = original_df.select_dtypes(include=[np.number])
if not numeric_df.empty:
# 添加统计列
stats_row = {
'statistic': 'mean',
**dict(zip(numeric_df.columns, np.mean(numeric_df.values, axis=0)))
}
# 创建新的DataFrame包含原始数据和统计信息
extended_df = original_df.copy()
stats_df = pd.DataFrame([stats_row])
final_df = pd.concat([extended_df, stats_df], ignore_index=True)
# 保存到CSV
final_df.to_csv(output_file, index=False, encoding='utf-8')
print(f"💾 已保存: {output_file}")
except Exception as e:
print(f"保存失败 {file_path}: {e}")
# 保存处理结果
save_processed_data(robust_results)
性能监控和基准测试
了解不同方法的性能差异对于优化代码很重要：
def performance_benchmark():
"""性能基准测试"""
# 创建大型测试数据
def create_large_test_data(filename, rows=10000, cols=10):
"""创建大型测试CSV文件"""
data = np.random.rand(rows, cols)
df = pd.DataFrame(data, columns=[f'col_{i}' for i in range(cols)])
df.to_csv(filename, index=False)
return filename
test_file = create_large_test_data('large_test.csv', 50000, 20)
methods = {
'pandas_only': lambda f: pd.read_csv(f).values,
'numpy_genfromtxt': lambda f: np.genfromtxt(f, delimiter=',', skip_header=1),
'chunked_pandas': lambda f: np.vstack([
chunk.values for chunk in pd.read_csv(f, chunksize=1000)
])
}
results = {}
for method_name, method_func in methods.items():
start_time = time.time()
try:
data = method_func(test_file)
end_time = time.time()
results[method_name] = {
'time': end_time - start_time,
'shape': data.shape if hasattr(data, 'shape') else 'N/A',
'success': True
}
print(f"⏱️ {method_name}: {results[method_name]['time']:.4f} 秒")
except Exception as e:
end_time = time.time()
results[method_name] = {
'time': end_time - start_time,
'error': str(e),
'success': False
}
print(f"💥 {method_name} 失败: {e}")
# 清理测试文件
if os.path.exists(test_file):
os.remove(test_file)
return results
# 运行性能测试
benchmark_results = performance_benchmark()
实际应用案例
销售数据分析系统
class SalesDataAnalyzer:
"""销售数据分析系统"""
def __init__(self, data_directory):
self.directory = data_directory
self.processed_data = {}
def load_sales_data(self, pattern="sales_*.csv"):
"""加载销售数据"""
files = glob.glob(os.path.join(self.directory, pattern))
for file_path in files:
try:
df = pd.read_csv(file_path)
# 提取关键指标
numeric_data = df.select_dtypes(include=[np.number]).values
if numeric_data.size > 0:
self.processed_data[file_path] = {
'total_sales': np.sum(numeric_data[:, 0]) if numeric_data.shape[1] > 0 else 0,
'total_quantity': np.sum(numeric_data[:, 1]) if numeric_data.shape[1] > 1 else 0,
'average_price': np.mean(numeric_data[:, 0]/numeric_data[:, 1]) if numeric_data.shape[1] > 1 and np.all(numeric_data[:, 1] != 0) else 0,
'sales_trend': np.polyfit(range(len(numeric_data)), numeric_data[:, 0], 1)[0] if len(numeric_data) > 1 else 0
}
except Exception as e:
print(f"加载销售数据失败 {file_path}: {e}")
def generate_report(self):
"""生成分析报告"""
if not self.processed_data:
print("没有可用数据")
return
print("📈 销售数据分析报告")
print("=" * 50)
total_sales = sum(data['total_sales'] for data in self.processed_data.values())
total_quantity = sum(data['total_quantity'] for data in self.processed_data.values())
print(f"总销售额: ¥{total_sales:,.2f}")
print(f"总销售量: {total_quantity:,}")
# 找出最佳表现文件
best_performing = max(self.processed_data.items(),
key=lambda x: x[1]['total_sales'])
print(f"最佳表现文件: {os.path.basename(best_performing[0])}")
print(f"销售额: ¥{best_performing[1]['total_sales']:,.2f}")
# 使用示例
analyzer = SalesDataAnalyzer(".")
analyzer.load_sales_data()
analyzer.generate_report()
财务数据批处理工具
def financial_data_processor(input_dir, output_dir):
"""财务数据批处理工具"""
# 确保输出目录存在
os.makedirs(output_dir, exist_ok=True)
# 查找所有财务相关的CSV文件
financial_files = glob.glob(os.path.join(input_dir, "financial_*.csv"))
summary_stats = {
'total_files': len(financial_files),
'processed_files': 0,
'failed_files': 0,
'total_records': 0
}
for file_path in financial_files:
try:
# 读取财务数据
df = pd.read_csv(file_path)
# 只保留数值列进行NumPy处理
numeric_df = df.select_dtypes(include=[np.number])
if not numeric_df.empty:
numeric_data = numeric_df.values
# 计算关键财务指标
financial_metrics = {
'mean': np.mean(numeric_data, axis=0),
'std': np.std(numeric_data, axis=0),
'percentiles': np.percentile(numeric_data, [25, 50, 75], axis=0),
'correlation_matrix': np.corrcoef(numeric_data.T) if numeric_data.shape[1] > 1 else np.array([[1.0]])
}
# 保存处理结果
base_name = os.path.splitext(os.path.basename(file_path))[0]
output_file = os.path.join(output_dir, f"{base_name}_analysis.npz")
np.savez_compressed(output_file,
raw_data=numeric_data,
metrics=financial_metrics,
column_names=numeric_df.columns.tolist())
summary_stats['processed_files'] += 1
summary_stats['total_records'] += numeric_data.shape[0]
print(f"✅ 财务分析完成: {file_path}")
else:
raise ValueError("没有数值数据可供分析")
except Exception as e:
summary_stats['failed_files'] += 1
print(f"❌ 财务分析失败 {file_path}: {e}")
# 输出汇总统计
print(f"\n💰 财务数据处理摘要:")
print(f" 总文件数: {summary_stats['total_files']}")
print(f" 成功处理: {summary_stats['processed_files']}")
print(f" 处理失败: {summary_stats['failed_files']}")
print(f" 总记录数: {summary_stats['total_records']:,}")
# 执行财务数据处理
financial_data_processor(".", "financial_analysis_output")
最佳实践和建议
通过以上实例，我们可以总结出一些使用NumPy批量处理CSV文件的最佳实践：
1. 选择合适的工具组合
# 推荐的工具组合
def recommended_approach(file_pattern):
"""推荐的CSV处理方法"""
files = glob.glob(file_pattern)
results = []
for file_path in files:
try:
# 使用pandas读取（处理能力强）
df = pd.read_csv(file_path)
# 使用NumPy进行数值计算（速度快）
numeric_data = df.select_dtypes(include=[np.number]).values
if numeric_data.size > 0:
# NumPy向量化操作
stats = {
'sum': np.sum(numeric_data, axis=0),
'mean': np.mean(numeric_data, axis=0),
'std': np.std(numeric_data, axis=0),
'min': np.min(numeric_data, axis=0),
'max': np.max(numeric_data, axis=0)
}
results.append({
'file': file_path,
'stats': stats,
'shape': numeric_data.shape
})
except Exception as e:
print(f"处理失败: {e}")
return results
2. 内存管理策略
def memory_efficient_processor(file_pattern, max_memory_mb=100):
"""内存高效的处理策略"""
files = glob.glob(file_pattern)
max_bytes = max_memory_mb * 1024 * 1024 # 转换为字节
for file_path in files:
try:
# 先检查文件大小
file_size = os.path.getsize(file_path)
if file_size > max_bytes:
print(f"⚠️ 大文件检测到: {file_path} ({file_size/1024/1024:.1f} MB)")
# 使用分块处理
chunk_processor(file_path)
else:
# 直接处理小文件
direct_processor(file_path)
except Exception as e:
print(f"处理失败: {e}")
def chunk_processor(file_path, chunk_size=1000):
"""分块处理器"""
print(f"📦 分块处理: {file_path}")
chunk_reader = pd.read_csv(file_path, chunksize=chunk_size)
chunk_count = 0
for chunk in chunk_reader:
numeric_chunk = chunk.select_dtypes(include=[np.number]).values
if numeric_chunk.size > 0:
# 在这里执行NumPy计算
chunk_mean = np.mean(numeric_chunk, axis=0)
print(f" 块 {chunk_count}: 平均值 = {chunk_mean}")
chunk_count += 1
def direct_processor(file_path):
"""直接处理器"""
print(f"⚡ 直接处理: {file_path}")
df = pd.read_csv(file_path)
numeric_data = df.select_dtypes(include=[np.number]).values
if numeric_data.size > 0:
overall_mean = np.mean(numeric_data)
print(f" 整体平均值: {overall_mean}")
3. 数据验证和质量控制
def data_quality_checker(file_path):
"""数据质量检查器"""
try:
df = pd.read_csv(file_path)
numeric_df = df.select_dtypes(include=[np.number])
if not numeric_df.empty:
numeric_data = numeric_df.values
quality_report = {
'total_cells': numeric_data.size,
'missing_cells': np.count_nonzero(np.isnan(numeric_data)),
'infinite_cells': np.count_nonzero(np.isinf(numeric_data)),
'valid_cells': np.count_nonzero(np.isfinite(numeric_data)),
'data_range': (np.nanmin(numeric_data), np.nanmax(numeric_data)),
'outliers': detect_outliers(numeric_data)
}
# 计算数据完整性百分比
completeness = quality_report['valid_cells'] / quality_report['total_cells'] * 100
print(f"🔍 数据质量报告 for {file_path}:")
print(f" 完整性: {completeness:.1f}%")
print(f" 缺失值: {quality_report['missing_cells']}")
print(f" 异常值: {len(quality_report['outliers'])}")
print(f" 数据范围: {quality_report['data_range'][0]:.2f} to {quality_report['data_range'][1]:.2f}")
return quality_report
except Exception as e:
print(f"质量检查失败: {e}")
return None
def detect_outliers(data, threshold=3):
"""使用Z-score检测异常值"""
if data.size == 0:
return []
z_scores = np.abs((data - np.mean(data)) / np.std(data))
outlier_indices = np.where(z_scores > threshold)
return list(zip(*outlier_indices)) if len(outlier_indices[0]) > 0 else []
# 执行质量检查
quality_result = data_quality_checker('sales_2023.csv')
扩展功能和集成
与数据库集成
def csv_to_database_processor(csv_files, db_connection_string):
"""CSV到数据库处理器"""
try:
import sqlite3
conn = sqlite3.connect(db_connection_string)
for file_path in csv_files:
try:
# 读取CSV
df = pd.read_csv(file_path)
# 转换为NumPy数组进行预处理
numeric_columns = df.select_dtypes(include=[np.number])
if not numeric_columns.empty:
# 数据清理
numeric_data = numeric_columns.values
cleaned_data = np.where(np.isnan(numeric_data), 0, numeric_data)
# 更新DataFrame
for i, col in enumerate(numeric_columns.columns):
df[col] = cleaned_data[:, i]
# 保存到数据库
table_name = os.path.splitext(os.path.basename(file_path))[0]
df.to_sql(table_name, conn, if_exists='replace', index=False)
print(f"💾 已保存到数据库: {table_name}")
except Exception as e:
print(f"数据库保存失败 {file_path}: {e}")
conn.close()
except ImportError:
print("⚠️ SQLite3未安装，请先安装数据库驱动")
except Exception as e:
print(f"数据库连接失败: {e}")
# 注意：这需要数据库支持，仅作示例展示
# csv_to_database_processor(['sales_2023.csv'], 'example.db')
Web服务集成
def create_processing_api():
"""创建简单的处理API"""
try:
from flask import Flask, request, jsonify
import json
app = Flask(__name__)
@app.route('/process-csv', methods=['POST'])
def process_csv_endpoint():
try:
# 这里应该是文件上传处理逻辑
# 为了简化，我们模拟处理过程
sample_data = np.random.rand(100, 5)
results = {
'mean': np.mean(sample_data, axis=0).tolist(),
'std': np.std(sample_data, axis=0).tolist(),
'shape': sample_data.shape
}
return jsonify({
'status': 'success',
'results': results
})
except Exception as e:
return jsonify({
'status': 'error',
'message': str(e)
}), 500
return app
except ImportError:
print("⚠️ Flask未安装，API功能不可用")
return None
# API应用示例（需要Flask支持）
# api_app = create_processing_api()
# if api_app:
# api_app.run(debug=True, port=5000)
总结和展望
通过本文的学习，我们掌握了使用Python NumPy批量处理CSV文件的核心技能：
- 基础操作：学会了使用NumPy读取和处理CSV数据的基本方法
- 批量处理：掌握了使用glob和pathlib进行多文件批量处理的技术
- 高级技巧：学习了数据清洗、性能优化、错误处理等高级技能
- 实际应用：通过具体的案例展示了在销售分析、财务处理等场景中的应用
NumPy的强大之处在于其高效的向量化操作能力，结合pandas的数据处理能力和Python的灵活性，我们可以构建出强大而高效的CSV批量处理系统。
在实际项目中，建议根据具体需求选择合适的方法：
- 对于小文件和简单处理，可以直接使用NumPy的loadtxt/genfromtxt
- 对于复杂的数据结构和大数据集，推荐使用pandas+NumPy的组合
- 对于超大数据集，考虑使用分块处理和内存优化策略
随着数据科学的发展，批量处理CSV文件的需求会越来越普遍。掌握这些技能不仅能够提高工作效率，还能为你在数据科学领域的进一步发展打下坚实基础。
记住，最好的学习方法是在实践中不断练习和完善。建议读者尝试用自己的数据集来练习这些技术，并根据实际需求进行调整和优化。
以上就是Python NumPy批量处理CSV文件数据的最佳实践教程的详细内容，更多关于Python NumPy处理CSV文件的资料请关注脚本之家其它相关文章！