使用PostgreSQL进行高效数据清洗的完全指南从识别异常数据到数据标准化一步步提升你的数据质量

威震华夏关云长

引言

在当今数据驱动的世界中，数据质量是组织成功的关键因素之一。低质量的数据会导致错误的决策、低效的运营和错失商机。数据清洗作为数据预处理的重要环节，是确保数据质量的关键步骤。PostgreSQL作为一款功能强大的开源关系型数据库管理系统，提供了丰富的工具和功能，使其成为进行高效数据清洗的理想选择。

本文将详细介绍如何使用PostgreSQL进行高效的数据清洗，从识别异常数据到数据标准化，帮助你一步步提升数据质量。我们将通过实际代码示例和最佳实践，展示如何充分利用PostgreSQL的功能来解决各种数据清洗挑战。

数据清洗的基本概念和流程

数据清洗是指识别并纠正（或删除）数据中的错误、不一致和不准确之处的过程。数据清洗的主要目标是提高数据质量，使其适合分析、报告和决策制定。

数据清洗的基本流程通常包括以下步骤：

1. 数据评估：了解数据的结构、内容和质量
2. 识别异常数据：发现缺失值、重复值、异常值和不一致数据
3. 处理异常数据：根据业务规则和数据特性，决定如何处理识别出的问题
4. 数据标准化：将数据转换为一致的格式和标准
5. 验证：确保清洗后的数据满足质量要求

PostgreSQL提供了强大的SQL功能、丰富的数据类型和内置函数，以及可扩展的架构，使其能够高效地执行这些数据清洗任务。

使用PostgreSQL识别异常数据

缺失值检测

缺失值是数据集中最常见的问题之一。在PostgreSQL中，我们可以使用多种方法来检测缺失值（NULL值）。

2. -- 检查表中每列的NULL值数量
3. SELECT
4.     column_name,
5.     COUNT(*) - COUNT(column_name) AS null_count
6. FROM
7.     your_table
8. CROSS JOIN
9.     (SELECT array_agg(attname) AS columns FROM pg_attribute WHERE attrelid = 'your_table'::regclass AND attnum > 0 AND NOT attisdropped) AS cols,
10.     unnest(columns) AS column_name
11. GROUP BY
12.     column_name;

复制代码

更简单的方法是针对特定列进行检查：

2. -- 检查特定列的NULL值
3. SELECT
4.     COUNT(*) AS total_rows,
5.     COUNT(column_name) AS non_null_rows,
6.     COUNT(*) - COUNT(column_name) AS null_rows,
7.     ROUND((COUNT(*) - COUNT(column_name)) * 100.0 / COUNT(*), 2) AS null_percentage
8. FROM
9.     your_table;

复制代码

对于大型表，我们可以使用采样方法来快速评估数据质量：

2. -- 使用TABLESAMPLE进行数据采样检查
3. SELECT
4.     column_name,
5.     COUNT(*) - COUNT(column_name) AS null_count
6. FROM
7.     your_table TABLESAMPLE BERNOULLI(10) -- 10%的随机采样
8. CROSS JOIN
9.     unnest(ARRAY['column1', 'column2', 'column3']) AS column_name
10. GROUP BY
11.     column_name;

复制代码

重复数据检测

重复数据是另一个常见的数据质量问题。PostgreSQL提供了多种方法来检测重复数据。

2. -- 检测完全重复的行
3. SELECT
4.     column1, column2, column3, COUNT(*) AS duplicate_count
5. FROM
6.     your_table
7. GROUP BY
8.     column1, column2, column3
9. HAVING
10.     COUNT(*) > 1;

复制代码

如果需要基于特定列检测重复数据：

2. -- 检测基于特定列的重复数据
3. SELECT
4.     duplicate_column, COUNT(*) AS duplicate_count
5. FROM
6.     your_table
7. GROUP BY
8.     duplicate_column
9. HAVING
10.     COUNT(*) > 1;

复制代码

要查看完整的重复记录，可以使用窗口函数：

2. -- 使用窗口函数查看重复记录
3. WITH duplicates AS (
4.     SELECT
5.         *,
6.         COUNT(*) OVER (PARTITION BY duplicate_column) AS duplicate_count
7.     FROM
8.         your_table
9. )
10. SELECT
11.     *
12. FROM
13.     duplicates
14. WHERE
15.     duplicate_count > 1
16. ORDER BY
17.     duplicate_column, id;

复制代码

异常值检测

异常值是显著偏离其他观测值的数据点。检测异常值的方法多种多样，下面是一些常用的统计方法。

使用标准差法检测数值异常：

2. -- 使用标准差法检测异常值（假设数据呈正态分布）
3. SELECT *
4. FROM your_table
5. WHERE numeric_column < (SELECT AVG(numeric_column) - 3 * STDDEV(numeric_column) FROM your_table)
6.    OR numeric_column > (SELECT AVG(numeric_column) + 3 * STDDEV(numeric_column) FROM your_table);

复制代码

使用四分位距（IQR）法检测异常值：

2. -- 使用四分位距（IQR）法检测异常值
3. WITH stats AS (
4.     SELECT
5.         PERCENTILE_CONT(0.25) WITHIN GROUP (ORDER BY numeric_column) AS q1,
6.         PERCENTILE_CONT(0.75) WITHIN GROUP (ORDER BY numeric_column) AS q3
7.     FROM
8.         your_table
9. )
10. SELECT *
11. FROM your_table, stats
12. WHERE numeric_column < (q1 - 1.5 * (q3 - q1))
13.    OR numeric_column > (q3 + 1.5 * (q3 - q1));

复制代码

对于文本数据，可以检测不符合特定模式的值：

2. -- 检测不符合特定模式的文本数据（例如，电子邮件格式）
3. SELECT *
4. FROM your_table
5. WHERE email_column !~ '^[A-Za-z0-9._%-]+@[A-Za-z0-9.-]+[.][A-Za-z]+$';

复制代码

不一致数据检测

不一致数据可能表现为格式不统一、编码不一致或违反业务规则等问题。

2. -- 检测日期格式不一致的数据
3. SELECT DISTINCT
4.     date_column,
5.     CASE
6.         WHEN date_column ~ '^\d{4}-\d{2}-\d{2}$' THEN 'YYYY-MM-DD'
7.         WHEN date_column ~ '^\d{2}/\d{2}/\d{4}$' THEN 'MM/DD/YYYY'
8.         WHEN date_column ~ '^\d{2}-\d{2}-\d{4}$' THEN 'MM-DD-YYYY'
9.         ELSE 'Unknown format'
10.     END AS detected_format
11. FROM
12.     your_table
13. WHERE
14.     date_column IS NOT NULL;

复制代码

2. -- 检测违反业务规则的数据（例如，年龄小于0或大于150）
3. SELECT *
4. FROM your_table
5. WHERE age_column < 0 OR age_column > 150;

复制代码

数据清洗技术

处理缺失值

处理缺失值的方法取决于数据特性和业务需求。常见的处理方法包括删除、替换和插补。

2. -- 删除包含NULL值的行
3. DELETE FROM your_table
4. WHERE column_name IS NULL;
6. -- 或者创建一个新表，不包含NULL值行
7. CREATE TABLE cleaned_table AS
8. SELECT * FROM your_table
9. WHERE column_name IS NOT NULL;

复制代码

2. -- 使用固定值替换NULL值
3. UPDATE your_table
4. SET column_name = 'default_value'
5. WHERE column_name IS NULL;
7. -- 使用列的平均值替换NULL值（数值列）
8. UPDATE your_table
9. SET numeric_column = COALESCE(numeric_column, (SELECT AVG(numeric_column) FROM your_table WHERE numeric_column IS NOT NULL))
10. WHERE numeric_column IS NULL;

复制代码

2. -- 使用前一个或后一个有效值填充NULL值（时间序列数据）
3. -- 首先确保表有适当的排序列，如时间戳
4. UPDATE your_table t1
5. SET column_name = (
6.     SELECT column_name
7.     FROM your_table t2
8.     WHERE t2.id < t1.id AND t2.column_name IS NOT NULL
9.     ORDER BY t2.id DESC
10.     LIMIT 1
11. )
12. WHERE t1.column_name IS NULL;

复制代码

对于更复杂的插补方法，可以使用PostgreSQL的扩展功能，如PL/R或PL/Python，实现线性插值、回归插值等高级方法。

处理重复数据

处理重复数据通常涉及删除重复项或合并重复记录。

2. -- 删除完全重复的行，保留每个组中的一个
3. DELETE FROM your_table
4. WHERE ctid NOT IN (
5.     SELECT min(ctid)
6.     FROM your_table
7.     GROUP BY column1, column2, column3
8. );
10. -- 注意：ctid是PostgreSQL的物理行标识符，每个表都有

复制代码

2. -- 创建一个新表，只包含不重复的记录
3. CREATE TABLE unique_table AS
4. SELECT DISTINCT ON (duplicate_column) *
5. FROM your_table
6. ORDER BY duplicate_column, id;

复制代码

对于需要合并重复记录的情况，可以使用聚合函数：

2. -- 合并重复记录，例如保留最新记录
3. WITH ranked AS (
4.     SELECT
5.         *,
6.         ROW_NUMBER() OVER (PARTITION BY duplicate_column ORDER BY update_timestamp DESC) AS rn
7.     FROM
8.         your_table
9. )
10. DELETE FROM your_table
11. WHERE id IN (SELECT id FROM ranked WHERE rn > 1);

复制代码

处理异常值

处理异常值的方法包括删除、替换和转换。

2. -- 删除异常值
3. DELETE FROM your_table
4. WHERE numeric_column < (SELECT AVG(numeric_column) - 3 * STDDEV(numeric_column) FROM your_table)
5.    OR numeric_column > (SELECT AVG(numeric_column) + 3 * STDDEV(numeric_column) FROM your_table);

复制代码

2. -- 将异常值替换为边界值（winsorization）
3. WITH stats AS (
4.     SELECT
5.         AVG(numeric_column) - 3 * STDDEV(numeric_column) AS lower_bound,
6.         AVG(numeric_column) + 3 * STDDEV(numeric_column) AS upper_bound
7.     FROM
8.         your_table
9. )
10. UPDATE your_table
11. SET numeric_column =
12.     CASE
13.         WHEN numeric_column < (SELECT lower_bound FROM stats) THEN (SELECT lower_bound FROM stats)
14.         WHEN numeric_column > (SELECT upper_bound FROM stats) THEN (SELECT upper_bound FROM stats)
15.         ELSE numeric_column
16.     END;

复制代码

2. -- 对数转换处理偏态分布中的异常值
3. UPDATE your_table
4. SET log_numeric_column = LOG(numeric_column + 1) -- 加1避免log(0)未定义
5. WHERE numeric_column > 0;

复制代码

数据转换和标准化

数据转换和标准化是确保数据一致性和可比性的关键步骤。

2. -- 文本数据标准化（例如，统一大小写）
3. UPDATE your_table
4. SET text_column = UPPER(text_column);
6. -- 或者使用首字母大写
7. UPDATE your_table
8. SET text_column = INITCAP(text_column);

复制代码

2. -- 去除文本数据中的多余空格
3. UPDATE your_table
4. SET text_column = TRIM(BOTH ' ' FROM text_column);
6. -- 替换多个空格为单个空格
7. UPDATE your_table
8. SET text_column = REGEXP_REPLACE(text_column, '[ ]{2,}', ' ', 'g');

复制代码

2. -- 标准化日期格式
3. UPDATE your_table
4. SET date_column = TO_DATE(date_column, 'YYYY-MM-DD')
5. WHERE date_column ~ '^\d{4}-\d{2}-\d{2}$';
7. UPDATE your_table
8. SET date_column = TO_DATE(date_column, 'MM/DD/YYYY')
9. WHERE date_column ~ '^\d{2}/\d{2}/\d{4}$';

复制代码

数据标准化和规范化

数据格式标准化

数据格式标准化确保相同类型的数据在整个数据库中具有一致的格式。

2. -- 电话号码标准化
3. UPDATE your_table
4. SET phone_column = REGEXP_REPLACE(phone_column, '[^0-9]', '', 'g');
6. -- 然后格式化为标准形式
7. UPDATE your_table
8. SET phone_column =
9.     CASE
10.         WHEN LENGTH(phone_column) = 10 THEN '(' || SUBSTRING(phone_column, 1, 3) || ') ' || SUBSTRING(phone_column, 4, 3) || '-' || SUBSTRING(phone_column, 7, 4)
11.         WHEN LENGTH(phone_column) = 11 AND SUBSTRING(phone_column, 1, 1) = '1' THEN '(' || SUBSTRING(phone_column, 2, 3) || ') ' || SUBSTRING(phone_column, 5, 3) || '-' || SUBSTRING(phone_column, 8, 4)
12.         ELSE phone_column -- 保持原样，如果不符合已知格式
13.     END;

复制代码

2. -- 地址标准化（简单示例）
3. UPDATE your_table
4. SET address_column = INITCAP(TRIM(BOTH ' ' FROM address_column));

复制代码

数据类型转换

确保数据以正确的类型存储对于数据质量和查询性能至关重要。

2. -- 将文本转换为数值类型
3. UPDATE your_table
4. SET numeric_column = CAST(text_numeric_column AS NUMERIC)
5. WHERE text_numeric_column ~ '^[0-9]+(\.[0-9]+)?$';
7. -- 将文本转换为日期类型
8. UPDATE your_table
9. SET date_column = TO_DATE(text_date_column, 'YYYY-MM-DD')
10. WHERE text_date_column ~ '^\d{4}-\d{2}-\d{2}$';

复制代码

2. -- 创建一个新表，使用正确的数据类型
3. CREATE TABLE typed_table AS
4. SELECT
5.     id,
6.     CAST(text_numeric_column AS NUMERIC) AS numeric_column,
7.     TO_DATE(text_date_column, 'YYYY-MM-DD') AS date_column,
8.     -- 其他列
9. FROM
10.     your_table;

复制代码

数据规范化

数据规范化是减少数据冗余和提高数据一致性的过程。

2. -- 示例：将非规范化的表转换为第一范式（1NF）
3. -- 假设原始表有一个包含多个值的列（如逗号分隔的标签）
5. -- 创建新表存储标签
6. CREATE TABLE tags (
7.     id SERIAL PRIMARY KEY,
8.     tag_name VARCHAR(100) UNIQUE NOT NULL
9. );
11. -- 创建关联表
12. CREATE TABLE item_tags (
13.     item_id INTEGER REFERENCES items(id),
14.     tag_id INTEGER REFERENCES tags(id),
15.     PRIMARY KEY (item_id, tag_id)
16. );
18. -- 提取唯一标签并插入tags表
19. INSERT INTO tags (tag_name)
20. SELECT DISTINCT UNNEST(STRING_TO_ARRAY(tags_column, ','))
21. FROM items
22. WHERE tags_column IS NOT NULL;
24. -- 填充关联表
25. INSERT INTO item_tags (item_id, tag_id)
26. SELECT
27.     i.id,
28.     t.id
29. FROM
30.     items i
31. CROSS JOIN
32.     UNNEST(STRING_TO_ARRAY(i.tags_column, ',')) AS tag_name
33. JOIN
34.     tags t ON t.tag_name = tag_name
35. WHERE
36.     i.tags_column IS NOT NULL;

复制代码

2. -- 示例：将表转换为第三范式（3NF）
3. -- 假设有一个包含员工和部门信息的表
5. -- 创建部门表
6. CREATE TABLE departments (
7.     id SERIAL PRIMARY KEY,
8.     department_name VARCHAR(100) UNIQUE NOT NULL,
9.     location VARCHAR(100)
10. );
12. -- 从员工表中提取唯一部门
13. INSERT INTO departments (department_name, location)
14. SELECT DISTINCT department_name, department_location
15. FROM employees
16. WHERE department_name IS NOT NULL;
18. -- 更新员工表，引用部门ID
19. ALTER TABLE employees ADD COLUMN department_id INTEGER;
21. UPDATE employees e
22. SET department_id = d.id
23. FROM departments d
24. WHERE e.department_name = d.department_name;
26. -- 移除原始部门列
27. ALTER TABLE employees DROP COLUMN department_name;
28. ALTER TABLE employees DROP COLUMN department_location;
30. -- 添加外键约束
31. ALTER TABLE employees ADD CONSTRAINT fk_department
32. FOREIGN KEY (department_id) REFERENCES departments(id);

复制代码

高级数据清洗技术

使用窗口函数进行数据清洗

窗口函数是PostgreSQL中强大的工具，特别适用于复杂的数据清洗任务。

2. -- 使用窗口函数填充时间序列中的缺失值
3. WITH numbered AS (
4.     SELECT
5.         date,
6.         value,
7.         ROW_NUMBER() OVER (ORDER BY date) AS rn
8.     FROM
9.         time_series_data
10. )
11. SELECT
12.     d1.date,
13.     COALESCE(
14.         d1.value,
15.         (
16.             SELECT d2.value
17.             FROM numbered d2
18.             WHERE d2.rn = (
19.                 SELECT MAX(d3.rn)
20.                 FROM numbered d3
21.                 WHERE d3.rn < d1.rn AND d3.value IS NOT NULL
22.             )
23.         )
24.     ) AS filled_value
25. FROM
26.     numbered d1
27. ORDER BY
28.     d1.date;

复制代码

2. -- 使用窗口函数识别数据中的突变点
3. WITH stats AS (
4.     SELECT
5.         date,
6.         value,
7.         LAG(value) OVER (ORDER BY date) AS prev_value,
8.         value - LAG(value) OVER (ORDER BY date) AS day_over_day_change,
9.         STDDEV(value) OVER (ORDER BY date ROWS BETWEEN 30 PRECEDING AND CURRENT ROW) AS rolling_stddev
10.     FROM
11.         time_series_data
12. )
13. SELECT
14.     date,
15.     value,
16.     prev_value,
17.     day_over_day_change,
18.     ABS(day_over_day_change) / rolling_stddev AS z_score
19. FROM
20.     stats
21. WHERE
22.     ABS(day_over_day_change) / rolling_stddev > 3 -- 3个标准差的变化
23. ORDER BY
24.     date;

复制代码

使用正则表达式进行复杂数据清洗

PostgreSQL的正则表达式功能非常强大，适用于复杂的数据清洗任务。

2. -- 从文本中提取特定模式的数据（例如，从地址中提取邮政编码）
3. UPDATE your_table
4. SET zip_code = REGEXP_REPLACE(address_column, '.*\b([0-9]{5})\b.*', '\1')
5. WHERE address_column ~ '\b[0-9]{5}\b';
7. -- 从混合格式的电话号码中提取数字
8. UPDATE your_table
9. SET clean_phone = REGEXP_REPLACE(phone_column, '[^0-9]', '', 'g');

复制代码

2. -- 标准化姓名格式（例如，将"Smith, John"转换为"John Smith"）
3. UPDATE your_table
4. SET full_name =
5.     CASE
6.         WHEN full_name ~ '^[A-Za-z-]+, [A-Za-z-]+$' THEN
7.             REGEXP_REPLACE(full_name, '^([A-Za-z-]+), ([A-Za-z-]+)$', '\2 \1')
8.         ELSE full_name
9.     END;

复制代码

2. -- 验证和标准化电子邮件地址
3. -- 首先检查电子邮件格式是否有效
4. SELECT *
5. FROM your_table
6. WHERE email_column !~ '^[A-Za-z0-9._%-]+@[A-Za-z0-9.-]+[.][A-Za-z]+$';
8. -- 标准化电子邮件地址（转换为小写）
9. UPDATE your_table
10. SET email_column = LOWER(TRIM(email_column));

复制代码

使用PostgreSQL扩展进行数据清洗

PostgreSQL的扩展生态系统提供了许多额外的功能，可以增强数据清洗能力。

2. -- 安装和使用fuzzystrmatch扩展进行模糊字符串匹配
3. -- 首先安装扩展
4. CREATE EXTENSION IF NOT EXISTS fuzzystrmatch;
6. -- 使用levenshtein距离查找相似的字符串
7. SELECT
8.     column1,
9.     column2,
10.     LEVENSHTEIN(column1, column2) AS distance
11. FROM
12.     your_table
13. WHERE
14.     LEVENSHTEIN(column1, column2) < 3; -- 距离小于3的字符串被认为是相似的

复制代码

2. -- 使用pg_trgm扩展进行更高级的文本匹配
3. -- 安装扩展
4. CREATE EXTENSION IF NOT EXISTS pg_trgm;
6. -- 使用相似度函数查找相似的字符串
7. SELECT
8.     column1,
9.     column2,
10.     SIMILARITY(column1, column2) AS similarity_score
11. FROM
12.     your_table
13. WHERE
14.     column1 % column2; -- %操作符表示相似度超过pg_trgm相似性阈值

复制代码

2. -- 使用tablefunc扩展进行数据透视
3. -- 安装扩展
4. CREATE EXTENSION IF NOT EXISTS tablefunc;
6. -- 创建交叉表（数据透视表）
7. SELECT *
8. FROM crosstab(
9.     'SELECT category_id, month, amount FROM sales ORDER BY 1,2',
10.     'SELECT DISTINCT month FROM sales ORDER BY 1'
11. ) AS ct(category_id INTEGER, jan NUMERIC, feb NUMERIC, mar NUMERIC, apr NUMERIC, may NUMERIC, jun NUMERIC,
12.        jul NUMERIC, aug NUMERIC, sep NUMERIC, oct NUMERIC, nov NUMERIC, dec NUMERIC);

复制代码

数据清洗的最佳实践

1. 始终备份原始数据

在进行任何数据清洗操作之前，始终创建原始数据的备份。

2. -- 创建原始表的备份
3. CREATE TABLE your_table_backup AS SELECT * FROM your_table;
5. -- 或者使用事务，以便在出现问题时可以回滚
6. BEGIN;
8. -- 执行数据清洗操作
9. UPDATE your_table SET column_name = TRIM(column_name);
11. -- 检查结果
12. SELECT COUNT(*) FROM your_table WHERE column_name LIKE ' %' OR column_name LIKE '% ';
14. -- 如果结果不符合预期，可以回滚
15. -- ROLLBACK;
17. -- 如果结果符合预期，提交事务
18. COMMIT;

复制代码

2. 使用事务确保数据一致性

对于复杂的数据清洗操作，使用事务可以确保操作的原子性。

2. BEGIN;
4. -- 执行多个相关的数据清洗操作
5. UPDATE customers SET email = LOWER(TRIM(email));
6. UPDATE orders SET status = UPPER(TRIM(status));
7. UPDATE products SET name = INITCAP(TRIM(name));
9. -- 验证结果
10. SELECT COUNT(*) FROM customers WHERE email <> LOWER(TRIM(email));
11. SELECT COUNT(*) FROM orders WHERE status <> UPPER(TRIM(status));
12. SELECT COUNT(*) FROM products WHERE name <> INITCAP(TRIM(name));
14. -- 如果所有计数都为0，则提交事务
15. COMMIT;
16. -- 否则回滚
17. -- ROLLBACK;

复制代码

3. 创建可重复的数据清洗脚本

将数据清洗操作组织成可重复的脚本，便于维护和重用。

2. -- 创建一个函数来封装数据清洗逻辑
3. CREATE OR REPLACE FUNCTION clean_customer_data()
4. RETURNS VOID AS $$
5. BEGIN
6.     -- 清洗电子邮件
7.     UPDATE customers SET email = LOWER(TRIM(email)) WHERE email IS NOT NULL;
8.    
9.     -- 标准化电话号码
10.     UPDATE customers SET phone = REGEXP_REPLACE(phone, '[^0-9]', '', 'g') WHERE phone IS NOT NULL;
11.    
12.     -- 清洗姓名
13.     UPDATE customers SET first_name = INITCAP(TRIM(first_name)) WHERE first_name IS NOT NULL;
14.     UPDATE customers SET last_name = INITCAP(TRIM(last_name)) WHERE last_name IS NOT NULL;
15.    
16.     -- 处理缺失值
17.     UPDATE customers SET middle_name = NULL WHERE TRIM(middle_name) = '';
18.    
19.     RAISE NOTICE 'Customer data cleaned successfully';
20. END;
21. $$ LANGUAGE plpgsql;
23. -- 执行函数
24. SELECT clean_customer_data();

复制代码

4. 记录数据清洗过程

维护数据清洗操作的记录，以便追踪和审计。

2. -- 创建数据清洗日志表
3. CREATE TABLE data_cleaning_log (
4.     id SERIAL PRIMARY KEY,
5.     table_name VARCHAR(100) NOT NULL,
6.     operation VARCHAR(100) NOT NULL,
7.     operation_time TIMESTAMP NOT NULL DEFAULT NOW(),
8.     rows_affected INTEGER,
9.     details TEXT
10. );
12. -- 创建一个函数来记录数据清洗操作
13. CREATE OR REPLACE FUNCTION log_cleaning_operation(
14.     p_table_name VARCHAR,
15.     p_operation VARCHAR,
16.     p_rows_affected INTEGER DEFAULT NULL,
17.     p_details TEXT DEFAULT NULL
18. ) RETURNS VOID AS $$
19. BEGIN
20.     INSERT INTO data_cleaning_log (table_name, operation, rows_affected, details)
21.     VALUES (p_table_name, p_operation, p_rows_affected, p_details);
22. END;
23. $$ LANGUAGE plpgsql;
25. -- 在数据清洗操作中使用日志函数
26. BEGIN;
28. -- 执行数据清洗操作
29. UPDATE customers SET email = LOWER(TRIM(email)) WHERE email IS NOT NULL;
31. -- 记录操作
32. SELECT log_cleaning_operation('customers', 'Standardize email format', ROW_COUNT);
34. COMMIT;

复制代码

5. 验证数据清洗结果

数据清洗后，验证结果以确保数据质量得到改善。

2. -- 创建数据质量检查函数
3. CREATE OR REPLACE FUNCTION validate_data_quality()
4. RETURNS TABLE(check_name VARCHAR, failed_rows INTEGER) AS $$
5. BEGIN
6.     RETURN QUERY
7.     -- 检查NULL值
8.     SELECT 'NULL emails' AS check_name, COUNT(*) AS failed_rows
9.     FROM customers WHERE email IS NULL
10.    
11.     UNION ALL
12.    
13.     -- 检查无效电子邮件格式
14.     SELECT 'Invalid email format', COUNT(*)
15.     FROM customers WHERE email !~ '^[A-Za-z0-9._%-]+@[A-Za-z0-9.-]+[.][A-Za-z]+$'
16.    
17.     UNION ALL
18.    
19.     -- 检查重复电子邮件
20.     SELECT 'Duplicate emails', COUNT(*) - COUNT(DISTINCT email)
21.     FROM customers WHERE email IS NOT NULL;
22. END;
23. $$ LANGUAGE plpgsql;
25. -- 执行数据质量检查
26. SELECT * FROM validate_data_quality();

复制代码

案例研究：使用PostgreSQL进行实际数据清洗项目

让我们通过一个实际案例来综合应用前面介绍的技术。假设我们有一个客户数据表，需要进行全面的数据清洗。

场景描述

我们有一个客户表customers_raw，包含以下字段：

• id: 客户ID
• first_name: 名
• last_name: 姓
• email: 电子邮件
• phone: 电话号码
• address: 地址
• city: 城市
• state: 州/省
• zip_code: 邮政编码
• registration_date: 注册日期
• last_purchase_date: 最后购买日期
• total_purchases: 总购买次数
• total_amount: 总购买金额

数据中存在各种质量问题，包括缺失值、重复记录、格式不一致、异常值等。

步骤1: 数据评估

首先，我们需要评估数据质量，识别问题。

2. -- 创建数据质量评估表
3. CREATE TABLE data_quality_assessment AS
4. SELECT
5.     'customers_raw' AS table_name,
6.     'first_name' AS column_name,
7.     COUNT(*) AS total_rows,
8.     COUNT(first_name) AS non_null_rows,
9.     COUNT(*) - COUNT(first_name) AS null_rows,
10.     ROUND((COUNT(*) - COUNT(first_name)) * 100.0 / COUNT(*), 2) AS null_percentage,
11.     COUNT(DISTINCT first_name) AS unique_values,
12.     COUNT(*) - COUNT(DISTINCT first_name) AS duplicate_values
13. FROM
14.     customers_raw
16. UNION ALL
18. SELECT
19.     'customers_raw' AS table_name,
20.     'last_name' AS column_name,
21.     COUNT(*) AS total_rows,
22.     COUNT(last_name) AS non_null_rows,
23.     COUNT(*) - COUNT(last_name) AS null_rows,
24.     ROUND((COUNT(*) - COUNT(last_name)) * 100.0 / COUNT(*), 2) AS null_percentage,
25.     COUNT(DISTINCT last_name) AS unique_values,
26.     COUNT(*) - COUNT(DISTINCT last_name) AS duplicate_values
27. FROM
28.     customers_raw
30. -- 为其他列重复此模式...
32. UNION ALL
34. SELECT
35.     'customers_raw' AS table_name,
36.     'email' AS column_name,
37.     COUNT(*) AS total_rows,
38.     COUNT(email) AS non_null_rows,
39.     COUNT(*) - COUNT(email) AS null_rows,
40.     ROUND((COUNT(*) - COUNT(email)) * 100.0 / COUNT(*), 2) AS null_percentage,
41.     COUNT(DISTINCT email) AS unique_values,
42.     COUNT(*) - COUNT(DISTINCT email) AS duplicate_values
43. FROM
44.     customers_raw;
46. -- 查看数据质量评估结果
47. SELECT * FROM data_quality_assessment ORDER BY null_percentage DESC;

复制代码

步骤2: 创建清洗后的表

为了避免直接修改原始数据，我们创建一个新表来存储清洗后的数据。

2. -- 创建清洗后的表结构
3. CREATE TABLE customers_clean (
4.     id SERIAL PRIMARY KEY,
5.     first_name VARCHAR(50),
6.     last_name VARCHAR(50),
7.     email VARCHAR(100) UNIQUE,
8.     phone VARCHAR(20),
9.     address VARCHAR(100),
10.     city VARCHAR(50),
11.     state VARCHAR(50),
12.     zip_code VARCHAR(20),
13.     registration_date DATE,
14.     last_purchase_date DATE,
15.     total_purchases INTEGER,
16.     total_amount NUMERIC(10,2)
17. );

复制代码

步骤3: 处理重复数据

2. -- 识别基于电子邮件的重复记录
3. WITH duplicate_emails AS (
4.     SELECT
5.         email,
6.         COUNT(*) AS duplicate_count
7.     FROM
8.         customers_raw
9.     WHERE
10.         email IS NOT NULL
11.     GROUP BY
12.         email
13.     HAVING
14.         COUNT(*) > 1
15. ),
16. ranked_customers AS (
17.     SELECT
18.         c.*,
19.         ROW_NUMBER() OVER (PARTITION BY c.email ORDER BY c.last_purchase_date DESC NULLS LAST, c.registration_date DESC) AS rn
20.     FROM
21.         customers_raw c
22.     JOIN
23.         duplicate_emails d ON c.email = d.email
24. )
25. -- 查看将被保留的记录
26. SELECT * FROM ranked_customers WHERE rn = 1;
28. -- 插入非重复记录和每个重复组中的最新记录
29. INSERT INTO customers_clean (
30.     first_name, last_name, email, phone, address, city, state, zip_code,
31.     registration_date, last_purchase_date, total_purchases, total_amount
32. )
33. SELECT
34.     first_name, last_name, email, phone, address, city, state, zip_code,
35.     registration_date, last_purchase_date, total_purchases, total_amount
36. FROM (
37.     SELECT
38.         c.*,
39.         ROW_NUMBER() OVER (PARTITION BY c.email ORDER BY c.last_purchase_date DESC NULLS LAST, c.registration_date DESC) AS rn
40.     FROM
41.         customers_raw c
42.     WHERE
43.         c.email IS NOT NULL
44. ) ranked
45. WHERE rn = 1;
47. -- 插入没有电子邮件的记录
48. INSERT INTO customers_clean (
49.     first_name, last_name, email, phone, address, city, state, zip_code,
50.     registration_date, last_purchase_date, total_purchases, total_amount
51. )
52. SELECT
53.     first_name, last_name, email, phone, address, city, state, zip_code,
54.     registration_date, last_purchase_date, total_purchases, total_amount
55. FROM
56.     customers_raw
57. WHERE
58.     email IS NULL;

复制代码

步骤4: 清洗和标准化数据

2. -- 清洗和标准化姓名
3. UPDATE customers_clean
4. SET first_name = INITCAP(TRIM(first_name))
5. WHERE first_name IS NOT NULL;
7. UPDATE customers_clean
8. SET last_name = INITCAP(TRIM(last_name))
9. WHERE last_name IS NOT NULL;
11. -- 标准化电子邮件
12. UPDATE customers_clean
13. SET email = LOWER(TRIM(email))
14. WHERE email IS NOT NULL;
16. -- 标准化电话号码
17. UPDATE customers_clean
18. SET phone = REGEXP_REPLACE(phone, '[^0-9]', '', 'g')
19. WHERE phone IS NOT NULL;
21. -- 格式化电话号码
22. UPDATE customers_clean
23. SET phone =
24.     CASE
25.         WHEN LENGTH(phone) = 10 THEN '(' || SUBSTRING(phone, 1, 3) || ') ' || SUBSTRING(phone, 4, 3) || '-' || SUBSTRING(phone, 7, 4)
26.         WHEN LENGTH(phone) = 11 AND SUBSTRING(phone, 1, 1) = '1' THEN '(' || SUBSTRING(phone, 2, 3) || ') ' || SUBSTRING(phone, 5, 3) || '-' || SUBSTRING(phone, 8, 4)
27.         ELSE phone
28.     END
29. WHERE phone IS NOT NULL;
31. -- 标准化地址
32. UPDATE customers_clean
33. SET address = INITCAP(TRIM(address))
34. WHERE address IS NOT NULL;
36. -- 标准化城市
37. UPDATE customers_clean
38. SET city = INITCAP(TRIM(city))
39. WHERE city IS NOT NULL;
41. -- 标准化州/省
42. UPDATE customers_clean
43. SET state = UPPER(TRIM(state))
44. WHERE state IS NOT NULL;
46. -- 标准化邮政编码
47. UPDATE customers_clean
48. SET zip_code = TRIM(zip_code)
49. WHERE zip_code IS NOT NULL;
51. -- 标准化日期
52. UPDATE customers_clean
53. SET registration_date = TO_DATE(registration_date, 'YYYY-MM-DD')
54. WHERE registration_date ~ '^\d{4}-\d{2}-\d{2}$';
56. UPDATE customers_clean
57. SET last_purchase_date = TO_DATE(last_purchase_date, 'YYYY-MM-DD')
58. WHERE last_purchase_date ~ '^\d{4}-\d{2}-\d{2}$';

复制代码

步骤5: 处理缺失值

2. -- 处理缺失的姓名
3. -- 如果first_name缺失但last_name存在，尝试从email中提取
4. UPDATE customers_clean
5. SET first_name = INITCAP(SPLIT_PART(SPLIT_PART(email, '@', 1), '.', 1))
6. WHERE first_name IS NULL AND email IS NOT NULL AND email ~ '@';
8. -- 如果last_name缺失但first_name存在，尝试从email中提取
9. UPDATE customers_clean
10. SET last_name = INITCAP(SPLIT_PART(SPLIT_PART(email, '@', 1), '.', 2))
11. WHERE last_name IS NULL AND email IS NOT NULL AND email ~ '@' AND email ~ '.';
13. -- 处理缺失的日期
14. -- 如果registration_date缺失，设置为当前日期
15. UPDATE customers_clean
16. SET registration_date = CURRENT_DATE
17. WHERE registration_date IS NULL;
19. -- 如果last_purchase_date缺失，设置为registration_date
20. UPDATE customers_clean
21. SET last_purchase_date = registration_date
22. WHERE last_purchase_date IS NULL AND registration_date IS NOT NULL;
24. -- 处理缺失的数值
25. -- 如果total_purchases缺失，设置为0
26. UPDATE customers_clean
27. SET total_purchases = 0
28. WHERE total_purchases IS NULL;
30. -- 如果total_amount缺失，设置为0
31. UPDATE customers_clean
32. SET total_amount = 0
33. WHERE total_amount IS NULL;

复制代码

步骤6: 处理异常值

2. -- 检测并处理total_purchases的异常值
3. WITH purchase_stats AS (
4.     SELECT
5.         AVG(total_purchases) AS avg_purchases,
6.         STDDEV(total_purchases) AS stddev_purchases
7.     FROM
8.         customers_clean
9. )
10. UPDATE customers_clean
11. SET total_purchases = (SELECT avg_purchases + 3 * stddev_purchases FROM purchase_stats)
12. WHERE total_purchases > (SELECT avg_purchases + 3 * stddev_purchases FROM purchase_stats);
14. -- 检测并处理total_amount的异常值
15. WITH amount_stats AS (
16.     SELECT
17.         AVG(total_amount) AS avg_amount,
18.         STDDEV(total_amount) AS stddev_amount
19.     FROM
20.         customers_clean
21. )
22. UPDATE customers_clean
23. SET total_amount = (SELECT avg_amount + 3 * stddev_amount FROM amount_stats)
24. WHERE total_amount > (SELECT avg_amount + 3 * stddev_amount FROM amount_stats);
26. -- 处理日期异常值（例如，last_purchase_date早于registration_date）
27. UPDATE customers_clean
28. SET last_purchase_date = registration_date
29. WHERE last_purchase_date < registration_date;

复制代码

步骤7: 验证清洗后的数据

2. -- 创建数据质量验证函数
3. CREATE OR REPLACE FUNCTION validate_customers_data()
4. RETURNS TABLE(check_name VARCHAR, failed_rows INTEGER) AS $$
5. BEGIN
6.     RETURN QUERY
7.     -- 检查NULL值
8.     SELECT 'NULL first_name' AS check_name, COUNT(*) AS failed_rows
9.     FROM customers_clean WHERE first_name IS NULL
10.    
11.     UNION ALL
12.    
13.     SELECT 'NULL last_name' AS check_name, COUNT(*) AS failed_rows
14.     FROM customers_clean WHERE last_name IS NULL
15.    
16.     UNION ALL
17.    
18.     SELECT 'NULL email' AS check_name, COUNT(*) AS failed_rows
19.     FROM customers_clean WHERE email IS NULL
20.    
21.     UNION ALL
22.    
23.     -- 检查无效电子邮件格式
24.     SELECT 'Invalid email format', COUNT(*)
25.     FROM customers_clean WHERE email IS NOT NULL AND email !~ '^[A-Za-z0-9._%-]+@[A-Za-z0-9.-]+[.][A-Za-z]+$'
26.    
27.     UNION ALL
28.    
29.     -- 检查重复电子邮件
30.     SELECT 'Duplicate emails', COUNT(*) - COUNT(DISTINCT email)
31.     FROM customers_clean WHERE email IS NOT NULL
32.    
33.     UNION ALL
34.    
35.     -- 检查日期异常
36.     SELECT 'Last purchase before registration', COUNT(*)
37.     FROM customers_clean WHERE last_purchase_date < registration_date
38.    
39.     UNION ALL
40.    
41.     -- 检查数值异常
42.     SELECT 'Negative total purchases', COUNT(*)
43.     FROM customers_clean WHERE total_purchases < 0
44.    
45.     UNION ALL
46.    
47.     SELECT 'Negative total amount', COUNT(*)
48.     FROM customers_clean WHERE total_amount < 0;
49. END;
50. $$ LANGUAGE plpgsql;
52. -- 执行数据质量验证
53. SELECT * FROM validate_customers_data();

复制代码

步骤8: 创建数据清洗报告

2. -- 创建数据清洗报告表
3. CREATE TABLE data_cleaning_report (
4.     id SERIAL PRIMARY KEY,
5.     cleaning_date TIMESTAMP NOT NULL DEFAULT NOW(),
6.     original_table VARCHAR(50) NOT NULL,
7.     cleaned_table VARCHAR(50) NOT NULL,
8.     original_rows INTEGER NOT NULL,
9.     cleaned_rows INTEGER NOT NULL,
10.     duplicates_removed INTEGER NOT NULL,
11.     null_values_handled INTEGER NOT NULL,
12.     outliers_handled INTEGER NOT NULL,
13.     validation_checks INTEGER NOT NULL,
14.     validation_failures INTEGER NOT NULL
15. );
17. -- 填充报告
18. INSERT INTO data_cleaning_report (
19.     original_table, cleaned_table, original_rows, cleaned_rows,
20.     duplicates_removed, null_values_handled, outliers_handled,
21.     validation_checks, validation_failures
22. )
23. SELECT
24.     'customers_raw' AS original_table,
25.     'customers_clean' AS cleaned_table,
26.     (SELECT COUNT(*) FROM customers_raw) AS original_rows,
27.     (SELECT COUNT(*) FROM customers_clean) AS cleaned_rows,
28.     (SELECT COUNT(*) FROM customers_raw) - (SELECT COUNT(DISTINCT email) FROM customers_raw WHERE email IS NOT NULL) -
29.     (SELECT COUNT(*) FROM customers_raw WHERE email IS NULL) AS duplicates_removed,
30.     (SELECT COUNT(*) FROM data_quality_assessment WHERE table_name = 'customers_raw' AND null_rows > 0) AS null_values_handled,
31.     (SELECT COUNT(*) FROM customers_raw WHERE total_purchases > (SELECT AVG(total_purchases) + 3 * STDDEV(total_purchases) FROM customers_raw) OR
32.         total_amount > (SELECT AVG(total_amount) + 3 * STDDEV(total_amount) FROM customers_raw)) AS outliers_handled,
33.     (SELECT COUNT(*) FROM validate_customers_data()) AS validation_checks,
34.     (SELECT SUM(failed_rows) FROM validate_customers_data()) AS validation_failures;
36. -- 查看报告
37. SELECT * FROM data_cleaning_report;

复制代码

结论

数据清洗是确保数据质量和可靠性的关键步骤，而PostgreSQL提供了丰富的工具和功能来支持高效的数据清洗过程。通过本文介绍的技术和最佳实践，你可以利用PostgreSQL的强大功能来识别异常数据、处理缺失值和重复数据、标准化数据格式，以及验证数据质量。

在实际应用中，数据清洗通常是一个迭代过程，需要根据具体的数据特性和业务需求进行调整。通过创建可重复的数据清洗脚本、记录清洗过程和验证结果，你可以建立一个可靠的数据清洗流程，持续提高数据质量。

记住，高质量的数据是做出准确决策和获得有价值洞察的基础。投资于数据清洗流程将为你的组织带来长期的回报，提高数据分析的准确性和可靠性。