数据清洗技术

一个简单的 ETL 管道，用于提取、转换和加载数据

1import pandas as pd
2from sqlalchemy import create_engine
3
4# 将数据提取为 pandas 的 DataFrame
5def extract_data(file_path):
6    return pd.read_csv(file_path)
7
8# 转换数据
9def transform_data(df):
10    # 处理缺失值
11    df.fillna(method='ffill', inplace=True)
12    # 数据变换（标准化）
13    df['scaled_column'] = (df['column'] - df['column'].mean()) / df['column'].std()
14    # 异常值检测与剔除
15    Q1 = df['column'].quantile(0.25)
16    Q3 = df['column'].quantile(0.75)
17    IQR = Q3 - Q1
18    df = df[~((df['column'] < (Q1 - 1.5 * IQR)) | (df['column'] > (Q3 + 1.5 * IQR)))]
19    return df
20
21# 加载数据
22def load_data(df, db_uri, table_name):
23    engine = create_engine(db_uri)
24    df.to_sql(table_name, engine, if_exists='replace', index=False)
25
26# 示例用法
27file_path = 'data.csv'
28db_uri = 'sqlite:///cleaned_data.db'
29table_name = 'cleaned_table'
30
31# 执行 ETL 流程
32data = extract_data(file_path)
33cleaned_data = transform_data(data)
34load_data(cleaned_data, db_uri, table_name)

部分数据清洗技术的 Python 伪代码示例

1# 数据标准化
2# 将所有日期格式统一为标准格式（例如 YYYY-MM-DD）
3def standardize_dates(data):
4    for record in data:
5        record['date'] = convert_to_standard_format(record['date'])
6    return data
7
8# 数据验证
9# 根据预定义规则校验数据（例如有效的邮政编码）
10def validate_data(data):
11    for record in data:
12        if not is_valid_postal_code(record['postal_code']):
13            raise ValueError("发现无效的邮政编码")
14        # 如需，验证电子邮件地址
15        if not is_valid_email(record['email']):
16            raise ValueError("发现无效的电子邮件地址")
17    return data
18
19# 数据填补
20# 使用对应列的平均值填充缺失值
21def impute_missing_values(data):
22    for column in data.columns:
23        if data[column].isnull().sum() > 0:
24            average_value = data[column].mean()
25            data[column].fillna(average_value, inplace=True)
26    return data
27
28# 数据转换
29# 将分类数据转换为数值形式
30# 以符合预定义的数据模式
31def transform_data(data):
32    for column in data.columns:
33        if data[column].dtype == 'object':  # 假设分类数据使用 'object' 类型
34            data[column] = convert_categorical_to_numerical(data[column])
35    return data
36
37# 主函数：处理数据并应用数据清洗技术
38def process_data(data):
39    data = standardize_dates(data)
40    data = validate_data(data)
41    data = impute_missing_values(data)
42    data = transform_data(data)
43    return data