脚本如何清洗整合采集数据

本文目录导读：

1. 数据清洗的核心流程
2. 常见数据清洗方法
3. 数据整合策略
4. 完整清洗脚本示例
5. 性能优化建议

我来详细介绍脚本清洗整合采集数据的常用方法和步骤。

数据清洗的核心流程

# 示例：使用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

1. 清洗前备份：保留原始数据
2. 记录清洗日志：追踪数据变化
3. 分步验证：每步清洗后检查结果
4. 规则文档化：记录清洗规则和决策
5. 自动化测试：编写数据质量测试用例

这样可以确保采集数据的质量和一致性,为后续分析打下良好基础。