[AI学习笔记]DeepSeek数据清洗流水线设计规范

# 示例代码：从MySQL数据库获取数据
import pandas as pd
import mysql.connector

def fetch_data_from_mysql(host, user, password, database, query):
    # 建立数据库连接
    conn = mysql.connector.connect(
        host=host,
        user=user,
        password=password,
        database=database
    )
    # 执行查询并获取数据
    data = pd.read_sql(query, conn)
    conn.close()
    return data

# 实例调用
mysql_data = fetch_data_from_mysql(
    host="your_host",
    user="your_user",
    password="your_password",
    database="your_database",
    query="SELECT * FROM your_table"
)

# 示例代码：数据格式统一与简单类型转换
import pandas as pd

def preprocess_data(data):
    # 复制数据以避免修改原始数据
    processed_data = data.copy()
    
    # 统一日期格式
    if 'date_column' in processed_data.columns:
        processed_data['date_column'] = pd.to_datetime(processed_data['date_column']).dt.strftime('%Y-%m-%d')
    
    # 字符串类型数字转数值类型
    if 'numeric_column' in processed_data.columns:
        processed_data['numeric_column'] = pd.to_numeric(processed_data['numeric_column'], errors='coerce')
    
    return processed_data

# 实例调用
preprocessed_data = preprocess_data(mysql_data)

# 示例代码：基于唯一标识符去重
def remove_duplicates(data, id_column):
    # 保留首次出现的记录，去除重复
    cleaned_data = data.drop_duplicates(subset=id_column, keep='first')
    return cleaned_data

# 实例调用
cleaned_data = remove_duplicates(preprocessed_data, 'id_column')

# 示例代码：缺失值填充（以均值填充为例）
def handle_missing_values(data, column, method='mean'):
    if method == 'mean':
        fill_value = data[column].mean()
    elif method == 'median':
        fill_value = data[column].median()
    elif method == 'mode':
        fill_value = data[column].mode()[0]
    else:
        fill_value = 0  # 默认填充0
    
    cleaned_data = data.fillna({column: fill_value})
    return cleaned_data

# 实例调用
cleaned_data = handle_missing_values(cleaned_data, 'numeric_column', 'mean')

# 示例代码：基于3σ原则检测并修复异常值
def detect_and_fix_outliers(data, column, method='cap_floor'):
    mean = data[column].mean()
    std = data[column].std()
    lower_bound = mean - 3 * std
    upper_bound = mean + 3 * std
    
    if method == 'cap_floor':
        # 将异常值限制在上下界内
        cleaned_data = data.copy()
        cleaned_data[column] = cleaned_data[column].clip(lower=lower_bound, upper=upper_bound)
    elif method == 'remove':
        # 删除包含异常值的记录
        cleaned_data = data[(data[column] >= lower_bound) & (data[column] <= upper_bound)]
    else:
        cleaned_data = data  # 不处理
    
    return cleaned_data

# 实例调用
cleaned_data = detect_and_fix_outliers(cleaned_data, 'numeric_column', 'cap_floor')

# 示例代码：数据完整性验证
def validate_data_integrity(data):
    # 检查缺失值
    missing_values = data.isnull().sum().sum()
    # 检查数据类型
    dtype_errors = []
    for col in data.columns:
        if 'date' in col.lower() and data[col].dtype != 'object':  # 假设日期列为字符串格式
            dtype_errors.append(col)
        elif 'numeric' in col.lower() and data[col].dtype not in ['int64', 'float64']:
            dtype_errors.append(col)
    
    return missing_values, dtype_errors

# 实例调用
missing_values, dtype_errors = validate_data_integrity(cleaned_data)
print(f"Missing values: {missing_values}")
print(f"Data type errors: {dtype_errors}")

# 示例代码：将数据存储到Parquet文件
def store_data_to_parquet(data, file_path):
    data.to_parquet(file_path, index=False)

# 实例调用
store_data_to_parquet(cleaned_data, 'cleaned_data.parquet')

# 示例命令：安装依赖库
pip install pandas numpy mysql-connector-python scikit-learn

# 示例：database_config.json
{
    "host": "your_host",
    "user": "your_user",
    "password": "your_password",
    "database": "your_database"
}

# 示例代码：run_pipeline.py
import json
import pandas as pd
from modules.data_collection import fetch_data_from_mysql
from modules.data_preprocessing import preprocess_data
from modules.data_cleaning import remove_duplicates, handle_missing_values, detect_and_fix_outliers
from modules.data_validation import validate_data_integrity
from modules.data_storage import store_data_to_parquet

def main():
    # 加载数据库配置
    with open('config/database_config.json', 'r') as f:
        db_config = json.load(f)
    
    # 数据收集
    query = "SELECT * FROM your_table"
    raw_data = fetch_data_from_mysql(
        host=db_config['host'],
        user=db_config['user'],
        password=db_config['password'],
        database=db_config['database'],
        query=query
    )
    
    # 数据预处理
    preprocessed_data = preprocess_data(raw_data)
    
    # 数据清洗
    cleaned_data = remove_duplicates(preprocessed_data, 'id_column')
    cleaned_data = handle_missing_values(cleaned_data, 'numeric_column', 'mean')
    cleaned_data = detect_and_fix_outliers(cleaned_data, 'numeric_column', 'cap_floor')
    
    # 数据验证
    missing_values, dtype_errors = validate_data_integrity(cleaned_data)
    if missing_values > 0 or len(dtype_errors) > 0:
        print("Data validation failed. Please check the data.")
        return
    
    # 数据存储
    store_data_to_parquet(cleaned_data, 'cleaned_data.parquet')
    print("Data cleaning pipeline completed successfully.")

if __name__ == "__main__":
    main()

# 示例代码：添加简单日志记录功能到主运行脚本
import logging

logging.basicConfig(
    filename='pipeline.log',
    level=logging.INFO,
    format='%(asctime)s - %(levelname)s - %(message)s'
)

def main():
    try:
        # ...（原有代码逻辑）
        logging.info("Data cleaning pipeline started.")
        # ...（原有代码逻辑）
        logging.info("Data cleaning pipeline completed successfully.")
    except Exception as e:
        logging.error(f"Data cleaning pipeline failed: {str(e)}")
        raise