脚本如何过滤数据库脏数据

本文目录导读：

1. Python脚本过滤示例
2. SQL脚本过滤（各种数据库）
3. Shell脚本批量处理
4. 通用Python ETL脚本
5. 最佳实践建议

Python脚本过滤示例

基础数据清洗脚本

import pandas as pd
import re
from datetime import datetime
def clean_data(df):
    # 1. 处理缺失值
    df = df.dropna(subset=['重要字段'])  # 删除关键字段为空的记录
    df['字段名'] = df['字段名'].fillna('默认值')
    # 2. 删除重复数据
    df = df.drop_duplicates(subset=['用户ID', '时间'], keep='last')
    # 3. 格式校验
    # 手机号格式
    df = df[df['手机号'].str.match(r'^1[3-9]\d{9}$', na=False)]
    # 邮箱格式
    df = df[df['邮箱'].str.contains(r'^[\w\.-]+@[\w\.-]+\.\w+$', na=False)]
    # 4. 范围校验
    df = df[(df['年龄'] >= 0) & (df['年龄'] <= 150)]
    df = df[df['金额'] > 0]
    # 5. 日期格式标准化
    df['日期'] = pd.to_datetime(df['日期'], errors='coerce')
    df = df.dropna(subset=['日期'])
    return df
# 使用示例
df = pd.read_sql('SELECT * FROM users', connection)
clean_df = clean_data(df)
clean_df.to_sql('users_clean', connection, if_exists='replace')

SQL脚本过滤（各种数据库）

MySQL

-- 创建清洗后的表
CREATE TABLE users_clean AS
SELECT * FROM users
WHERE 
    -- 去除空值
    user_id IS NOT NULL
    AND phone IS NOT NULL
    -- 格式验证
    AND phone REGEXP '^1[3-9][0-9]{9}$'
    AND email REGEXP '^[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Za-z]{2,}$'
    -- 范围验证
    AND age BETWEEN 0 AND 150
    AND amount >= 0
    -- 去重（保留最新记录）
    AND id IN (
        SELECT MAX(id) 
        FROM users 
        GROUP BY user_id, date
    );
-- 使用窗口函数去重
WITH dedup AS (
    SELECT *,
        ROW_NUMBER() OVER (
            PARTITION BY user_id, order_date 
            ORDER BY updated_at DESC
        ) AS rn
    FROM users
)
SELECT * FROM dedup WHERE rn = 1;

PostgreSQL

-- 使用窗口函数处理脏数据
WITH cleaned AS (
    SELECT 
        *,
        -- 标记无效记录
        CASE 
            WHEN email ~ '^[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Za-z]{2,}$' 
            THEN TRUE ELSE FALSE 
        END AS is_valid_email,
        -- 去除特殊字符
        regexp_replace(phone, '[^0-9]', '', 'g') AS clean_phone
    FROM users
)
SELECT * FROM cleaned 
WHERE 
    is_valid_email = TRUE
    AND LENGTH(clean_phone) = 11
    AND age > 0 AND age < 150;

Shell脚本批量处理

#!/bin/bash
# 数据库连接参数
DB_HOST="localhost"
DB_USER="root"
DB_PASS="password"
DB_NAME="test"
# 定义数据质量检查函数
check_data_quality() {
    mysql -h $DB_HOST -u $DB_USER -p$DB_PASS $DB_NAME << EOF
    -- 检查空值率
    SELECT 
        '空值统计' as check_type,
        COUNT(*) as total_rows,
        SUM(CASE WHEN name IS NULL THEN 1 ELSE 0 END) as null_names,
        SUM(CASE WHEN phone IS NULL THEN 1 ELSE 0 END) as null_phones
    FROM users;
    -- 检查重复率
    SELECT 
        '重复统计' as check_type,
        COUNT(*) as total,
        COUNT(DISTINCT user_id) as unique_users
    FROM users;
    -- 检查异常值
    SELECT 
        '异常值统计' as check_type,
        COUNT(*) as invalid_ages
    FROM users 
    WHERE age < 0 OR age > 150;
EOF
}
# 执行数据清洗
clean_dirty_data() {
    mysql -h $DB_HOST -u $DB_USER -p$DB_PASS $DB_NAME << EOF
    -- 创建备份表
    CREATE TABLE IF NOT EXISTS users_backup AS SELECT * FROM users;
    -- 删除无效记录
    DELETE FROM users 
    WHERE 
        name IS NULL 
        OR phone IS NULL 
        OR LENGTH(phone) != 11
        OR age < 0 OR age > 150;
    -- 去重
    DELETE t1 FROM users t1
    INNER JOIN users t2 
    WHERE 
        t1.id < t2.id 
        AND t1.user_id = t2.user_id;
    -- 更新格式
    UPDATE users 
    SET phone = REGEXP_REPLACE(phone, '[^0-9]', '')
    WHERE phone REGEXP '[^0-9]';
EOF
}
# 主流程
echo "开始数据质量检查..."
check_data_quality
echo "开始数据清洗..."
clean_dirty_data
echo "清洗完成！"

通用Python ETL脚本

import pymysql
import logging
from datetime import datetime
class DataCleaner:
    def __init__(self, db_config):
        self.connection = pymysql.connect(**db_config)
        logging.basicConfig(level=logging.INFO)
    def find_dirty_data(self, table, rules):
        """查找脏数据"""
        conditions = []
        for field, rule in rules.items():
            if rule['type'] == 'not_null':
                conditions.append(f"{field} IS NULL")
            elif rule['type'] == 'regex':
                conditions.append(f"{field} NOT REGEXP '{rule['pattern']}'")
            elif rule['type'] == 'range':
                conditions.append(f"{field} {rule['operator']} {rule['value']}")
        where_clause = " OR ".join(conditions)
        query = f"SELECT * FROM {table} WHERE {where_clause}"
        with self.connection.cursor() as cursor:
            cursor.execute(query)
            return cursor.fetchall()
    def clean_data(self, table, rules):
        """清洗数据"""
        # 记录脏数据到日志表
        dirty_records = self.find_dirty_data(table, rules)
        self.log_dirty_data(table, dirty_records)
        # 执行清洗操作
        with self.connection.cursor() as cursor:
            for field, rule in rules.items():
                if rule['action'] == 'delete':
                    cursor.execute(f"""
                        DELETE FROM {table} 
                        WHERE {field} IS NULL
                    """)
                elif rule['action'] == 'update':
                    cursor.execute(f"""
                        UPDATE {table} 
                        SET {field} = '{rule['default']}'
                        WHERE {field} IS NULL
                    """)
        self.connection.commit()
    def log_dirty_data(self, table, records):
        """记录脏数据日志"""
        log_query = """
            INSERT INTO data_clean_log 
            (table_name, record_id, dirty_fields, clean_time)
            VALUES (%s, %s, %s, %s)
        """
        with self.connection.cursor() as cursor:
            for record in records:
                cursor.execute(log_query, (
                    table,
                    record['id'],
                    'dirty_fields_detected',
                    datetime.now()
                ))
        self.connection.commit()
# 使用示例
rules = {
    'phone': {
        'type': 'regex',
        'pattern': '^1[3-9][0-9]{9}$',
        'action': 'delete'
    },
    'email': {
        'type': 'regex',
        'pattern': '^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}$',
        'action': 'update',
        'default': 'unknown@example.com'
    },
    'age': {
        'type': 'range',
        'operator': 'NOT BETWEEN',
        'value': '0 AND 150',
        'action': 'delete'
    }
}
cleaner = DataCleaner(db_config)
cleaner.clean_data('users', rules)

最佳实践建议

脏数据分类处理

# 脏数据分类
DIRTY_TYPES = {
    'missing': '缺失值',
    'duplicate': '重复数据', 
    'invalid_format': '格式错误',
    'out_of_range': '超出范围',
    'inconsistent': '数据不一致',
    'abnormal': '异常值'
}
# 分级处理策略
CLEAN_STRATEGIES = {
    'critical': ['delete', 'report'],  # 关键字段错误直接删除
    'warning': ['update', 'log'],      # 非关键字段更新并记录
    'info': ['correct', 'notify']      # 小错误自动修正并通知
}

自动化监控脚本

# 定时任务示例（cron）
def schedule_clean():
    """每天凌晨执行数据清洗"""
    # 1. 备份数据
    backup_data()
    # 2. 执行清洗
    clean_dirty_data()
    # 3. 生成报告
    generate_clean_report()
    # 4. 发送告警
    if has_critical_issues:
        send_alert('数据质量问题严重')
# crontab配置
# 0 3 * * * /usr/bin/python3 /scripts/clean_data.py

这些脚本可以根据您的具体需求进行调整和组合使用,建议先在小规模数据上测试，确认逻辑正确后再应用于生产环境。