本文目录导读：
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
这些脚本可以根据您的具体需求进行调整和组合使用,建议先在小规模数据上测试，确认逻辑正确后再应用于生产环境。
