本文目录导读：
我来详细介绍脚本清洗整合采集数据的常用方法和步骤。
数据清洗的核心流程
# 示例：使用Pandas进行数据清洗
import pandas as pd
import numpy as np
import re
def clean_collected_data(df):
"""数据清洗主函数"""
# 1. 重复数据处理
df = df.drop_duplicates()
df = df.drop_duplicates(subset=['关键字段'], keep='first')
# 2. 缺失值处理
# 删除缺失率超过50%的列
df = df.dropna(thresh=len(df)*0.5, axis=1)
# 填充数值列
df['数值列'] = df['数值列'].fillna(df['数值列'].median())
# 填充文本列
df['文本列'] = df['文本列'].fillna('未知')
# 3. 数据类型转换
df['日期'] = pd.to_datetime(df['日期'], errors='coerce')
df['金额'] = pd.to_numeric(df['金额'], errors='coerce')
return df
常见数据清洗方法
文本数据清洗
import re
def clean_text_data(text):
"""清洗文本数据"""
# 去除HTML标签
text = re.sub(r'<[^>]+>', '', text)
# 去除特殊字符
text = re.sub(r'[^\w\s]', '', text)
# 去除多余空格
text = ' '.join(text.split())
# 统一大小写
text = text.lower()
# 去除停用词（示例）
stop_words = ['的', '了', '在', '是', '我']
text = ' '.join([word for word in text.split() if word not in stop_words])
return text
格式化清洗
def format_fields(df):
"""格式化字段"""
# 统一电话号码格式
df['电话'] = df['电话'].str.replace(r'[-\s+]', '', regex=True)
df['电话'] = df['电话'].apply(lambda x: f"+86 {x}" if x.startswith('1') else x)
# 统一日期格式
df['日期'] = pd.to_datetime(df['日期']).dt.strftime('%Y-%m-%d')
# 标准化地址
df['地址'] = df['地址'].str.replace(r'[\s]', '', regex=True)
return df
异常值处理
def handle_outliers(df, column, method='iqr'):
"""处理异常值"""
if method == 'iqr':
Q1 = df[column].quantile(0.25)
Q3 = df[column].quantile(0.75)
IQR = Q3 - Q1
lower_bound = Q1 - 1.5 * IQR
upper_bound = Q3 + 1.5 * IQR
df[column] = df[column].clip(lower_bound, upper_bound)
elif method == 'zscore':
from scipy import stats
z_scores = np.abs(stats.zscore(df[column]))
df[column][z_scores > 3] = df[column].median()
return df
数据整合策略
多源数据合并
def merge_multiple_sources():
"""合并多个数据源"""
# 读取不同来源的数据
df_csv = pd.read_csv('source1.csv')
df_json = pd.read_json('source2.json')
df_db = pd.read_sql('SELECT * FROM table', connection)
# 横向合并（列合并）
merged_horizontal = pd.concat([df_csv, df_json], axis=1)
# 纵向合并（行合并）
merged_vertical = pd.concat([df_csv, df_db], ignore_index=True)
# 根据关键字段合并（类似SQL JOIN）
merged_on_key = pd.merge(
df_csv, df_json,
on='user_id',
how='outer',
suffixes=('_left', '_right')
)
return merged_on_key
字段映射与统一
def unify_fields(df):
"""统一字段名称和类型"""
# 字段名映射
field_mapping = {
'name': '姓名',
'age': '年龄',
'tel': '电话',
'email': '邮箱'
}
df = df.rename(columns=field_mapping)
# 值映射
gender_mapping = {
'M': '男',
'F': '女',
'male': '男',
'female': '女',
'1': '男',
'0': '女'
}
if '性别' in df.columns:
df['性别'] = df['性别'].map(gender_mapping)
return df
完整清洗脚本示例
import pandas as pd
import numpy as np
from datetime import datetime
import logging
class DataCleaner:
"""数据清洗器"""
def __init__(self, data_path):
self.data_path = data_path
self.df = None
self.clean_log = []
def load_data(self):
"""加载数据"""
try:
self.df = pd.read_csv(self.data_path)
self.clean_log.append(f"加载数据: {len(self.df)} 行")
except Exception as e:
logging.error(f"数据加载失败: {e}")
raise
def clean_data(self):
"""执行清洗流程"""
# 1. 删除完全重复的行
before = len(self.df)
self.df = self.df.drop_duplicates()
self.clean_log.append(f"删除重复行: {before - len(self.df)} 行")
# 2. 处理缺失值
for col in self.df.columns:
null_count = self.df[col].isnull().sum()
if null_count > 0:
if self.df[col].dtype in ['object']:
self.df[col] = self.df[col].fillna('未知')
else:
self.df[col] = self.df[col].fillna(self.df[col].median())
self.clean_log.append(f"填充 {col} 缺失值: {null_count}")
# 3. 格式化日期
for col in self.df.select_dtypes(include=['object']).columns:
if '日期' in col or 'time' in col.lower():
try:
self.df[col] = pd.to_datetime(self.df[col])
except:
pass
# 4. 清理文本
for col in self.df.select_dtypes(include=['object']).columns:
self.df[col] = self.df[col].str.strip()
self.df[col] = self.df[col].str.replace(r'\s+', ' ', regex=True)
return self.clean_log
def integrate_data(self, other_data):
"""整合其他数据"""
# 检查共同字段
common_columns = list(set(self.df.columns) & set(other_data.columns))
if 'id' in common_columns:
self.df = pd.merge(self.df, other_data, on='id', how='left')
else:
self.df = pd.concat([self.df, other_data], ignore_index=True)
return self.df
def validate_data(self):
"""验证数据质量"""
validation_result = {
'total_rows': len(self.df),
'total_columns': len(self.df.columns),
'duplicates': len(self.df[self.df.duplicated()]),
'missing_values': self.df.isnull().sum().to_dict(),
'data_types': self.df.dtypes.to_dict()
}
return validation_result
def save_clean_data(self, output_path):
"""保存清洗后的数据"""
self.df.to_csv(output_path, index=False, encoding='utf-8-sig')
self.clean_log.append(f"数据已保存: {output_path}")
# 使用示例
if __name__ == "__main__":
cleaner = DataCleaner("collected_data.csv")
cleaner.load_data()
print("清洗日志:", cleaner.clean_data())
print("数据验证:", cleaner.validate_data())
cleaner.save_clean_data("cleaned_data.csv")
性能优化建议
分批次处理大数据
def process_large_file(file_path, chunk_size=10000):
"""分块处理大文件"""
chunks = []
for chunk in pd.read_csv(file_path, chunksize=chunk_size):
# 清洗每个数据块
cleaned_chunk = clean_chunk(chunk)
chunks.append(cleaned_chunk)
return pd.concat(chunks, ignore_index=True)
并行处理
from multiprocessing import Pool
def parallel_clean(data_list):
"""并行清洗"""
with Pool(processes=4) as pool:
results = pool.map(clean_function, data_list)
return results
- 清洗前备份：保留原始数据
- 记录清洗日志：追踪数据变化
- 分步验证：每步清洗后检查结果
- 规则文档化：记录清洗规则和决策
- 自动化测试：编写数据质量测试用例
这样可以确保采集数据的质量和一致性,为后续分析打下良好基础。
