# 创建数据质量规则表
CREATE TABLE DATA_QUALITY_RULES (
RULE_ID INTEGER NOT NULL GENERATED ALWAYS AS IDENTITY,
TABLE_NAME VARCHAR(100) NOT NULL,
COLUMN_NAME VARCHAR(100) NOT NULL,
RULE_TYPE VARCHAR(50) NOT NULL,
RULE_DESC VARCHAR(500) NOT NULL,
RULE_SQL VARCHAR(1000) NOT NULL,
THRESHOLD DECIMAL(5, 2) DEFAULT 100.00,
IS_ACTIVE CHAR(1) DEFAULT ‘Y’,
CREATE_TIME TIMESTAMP NOT NULL DEFAULT CURRENT TIMESTAMP,
CONSTRAINT PK_DATA_QUALITY_RULES PRIMARY KEY (RULE_ID)
);

INSERT INTO DATA_QUALITY_RULES (TABLE_NAME, COLUMN_NAME, RULE_TYPE, RULE_DESC, RULE_SQL)
VALUES
(‘CUSTOMER’, ‘CUSTOMER_ID’, ‘NOT_NULL’, ‘客户ID不能为空’, ‘SELECT COUNT(*) FROM CUSTOMER WHERE CUSTOMER_ID IS NULL’),
(‘CUSTOMER’, ‘EMAIL’, ‘FORMAT’, ‘邮箱格式检查’, ‘SELECT COUNT(*) FROM CUSTOMER WHERE EMAIL NOT LIKE ”%@%.%”’),
(‘ORDERS’, ‘ORDER_AMOUNT’, ‘RANGE’, ‘订单金额范围检查’, ‘SELECT COUNT(*) FROM ORDERS WHERE ORDER_AMOUNT < 0 OR ORDER_AMOUNT > 1000000′),
(‘ORDERS’, ‘ORDER_DATE’, ‘NOT_NULL’, ‘订单日期不能为空’, ‘SELECT COUNT(*) FROM ORDERS WHERE ORDER_DATE IS NULL’);

# 执行数据质量检查
CREATE OR REPLACE PROCEDURE SP_CHECK_DATA_QUALITY()
LANGUAGE SQL
BEGIN
DECLARE v_rule_id INTEGER;
DECLARE v_table_name VARCHAR(100);
DECLARE v_column_name VARCHAR(100);
DECLARE v_rule_type VARCHAR(50);
DECLARE v_rule_sql VARCHAR(1000);
DECLARE v_error_count INTEGER;
DECLARE v_threshold DECIMAL(5, 2);

FOR rule_cursor AS
SELECT RULE_ID, TABLE_NAME, COLUMN_NAME, RULE_TYPE, RULE_SQL, THRESHOLD
FROM DATA_QUALITY_RULES
WHERE IS_ACTIVE = ‘Y’
DO
SET v_rule_id = rule_cursor.RULE_ID;
SET v_table_name = rule_cursor.TABLE_NAME;
SET v_column_name = rule_cursor.COLUMN_NAME;
SET v_rule_type = rule_cursor.RULE_TYPE;
SET v_rule_sql = rule_cursor.RULE_SQL;
SET v_threshold = rule_cursor.THRESHOLD;

— 执行规则SQL
PREPARE stmt FROM v_rule_sql;
EXECUTE stmt INTO v_error_count;

— 记录检查结果
INSERT INTO DATA_QUALITY_RESULTS (
RULE_ID, TABLE_NAME, COLUMN_NAME, RULE_TYPE,
ERROR_COUNT, CHECK_TIME
) VALUES (
v_rule_id, v_table_name, v_column_name, v_rule_type,
v_error_count, CURRENT TIMESTAMP
);
END FOR;
END;

# 创建数据质量结果表
CREATE TABLE DATA_QUALITY_RESULTS (
RESULT_ID BIGINT NOT NULL GENERATED ALWAYS AS IDENTITY,
RULE_ID INTEGER NOT NULL,
TABLE_NAME VARCHAR(100) NOT NULL,
COLUMN_NAME VARCHAR(100) NOT NULL,
RULE_TYPE VARCHAR(50) NOT NULL,
ERROR_COUNT INTEGER NOT NULL,
CHECK_TIME TIMESTAMP NOT NULL DEFAULT CURRENT TIMESTAMP,
CONSTRAINT PK_DATA_QUALITY_RESULTS PRIMARY KEY (RESULT_ID)
);

# 数据质量监控脚本
#!/bin/bash
# data_quality_monitor.sh

DBNAME=FGEDB

db2 connect to $DBNAME

# 执行数据质量检查
db2 “CALL SP_CHECK_DATA_QUALITY()”

# 查询检查结果
db2 “SELECT
TABLE_NAME,
COLUMN_NAME,
RULE_TYPE,
ERROR_COUNT,
CHECK_TIME
FROM DATA_QUALITY_RESULTS
WHERE CHECK_TIME > CURRENT TIMESTAMP – 1 HOUR
AND ERROR_COUNT > 0”

db2 connect reset

# 发送告警邮件
if [ -s /tmp/quality_alert.txt ]; then
mail -s “Data Quality Alert” admin@example.com < /tmp/quality_alert.txt fi

# 创建数据分类表
CREATE TABLE DATA_CLASSIFICATION (
CLASSIFICATION_ID INTEGER NOT NULL GENERATED ALWAYS AS IDENTITY,
TABLE_NAME VARCHAR(100) NOT NULL,
COLUMN_NAME VARCHAR(100) NOT NULL,
DATA_TYPE VARCHAR(50) NOT NULL,
SENSITIVITY_LEVEL VARCHAR(20) NOT NULL,
DESCRIPTION VARCHAR(500),
CREATE_TIME TIMESTAMP NOT NULL DEFAULT CURRENT TIMESTAMP,
CONSTRAINT PK_DATA_CLASSIFICATION PRIMARY KEY (CLASSIFICATION_ID)
);

INSERT INTO DATA_CLASSIFICATION (TABLE_NAME, COLUMN_NAME, DATA_TYPE, SENSITIVITY_LEVEL, DESCRIPTION)
VALUES
(‘CUSTOMER’, ‘CUSTOMER_NAME’, ‘PERSONAL’, ‘MEDIUM’, ‘客户姓名’),
(‘CUSTOMER’, ‘ID_CARD’, ‘PERSONAL’, ‘HIGH’, ‘身份证号’),
(‘CUSTOMER’, ‘PHONE’, ‘PERSONAL’, ‘MEDIUM’, ‘手机号码’),
(‘CUSTOMER’, ‘EMAIL’, ‘PERSONAL’, ‘LOW’, ‘电子邮箱’),
(‘CUSTOMER’, ‘BANK_ACCOUNT’, ‘FINANCIAL’, ‘HIGH’, ‘银行账号’),
(‘ORDERS’, ‘ORDER_AMOUNT’, ‘FINANCIAL’, ‘MEDIUM’, ‘订单金额’);

# 查询敏感数据
SELECT
TABLE_NAME,
COLUMN_NAME,
DATA_TYPE,
SENSITIVITY_LEVEL
FROM DATA_CLASSIFICATION
WHERE SENSITIVITY_LEVEL IN (‘HIGH’, ‘MEDIUM’)
ORDER BY SENSITIVITY_LEVEL DESC;

# 创建数据脱敏函数
CREATE OR REPLACE FUNCTION FN_MASK_ID_CARD(p_id_card VARCHAR(18))
RETURNS VARCHAR(18)
LANGUAGE SQL
BEGIN
RETURN SUBSTR(p_id_card, 1, 6) || ‘********’ || SUBSTR(p_id_card, 15, 4);
END;

CREATE OR REPLACE FUNCTION FN_MASK_PHONE(p_phone VARCHAR(11))
RETURNS VARCHAR(11)
LANGUAGE SQL
BEGIN
RETURN SUBSTR(p_phone, 1, 3) || ‘****’ || SUBSTR(p_phone, 8, 4);
END;

CREATE OR REPLACE FUNCTION FN_MASK_BANK_ACCOUNT(p_account VARCHAR(20))
RETURNS VARCHAR(20)
LANGUAGE SQL
BEGIN
RETURN SUBSTR(p_account, 1, 4) || ‘************’ || SUBSTR(p_account, 17, 4);
END;

# 创建脱敏视图
CREATE VIEW CUSTOMER_MASKED AS
SELECT
CUSTOMER_ID,
CUSTOMER_NAME,
FN_MASK_ID_CARD(ID_CARD) AS ID_CARD,
FN_MASK_PHONE(PHONE) AS PHONE,
EMAIL,
FN_MASK_BANK_ACCOUNT(BANK_ACCOUNT) AS BANK_ACCOUNT
FROM CUSTOMER;

# 使用脱敏视图查询
SELECT * FROM CUSTOMER_MASKED;

# 创建归档表
CREATE TABLE ORDERS_ARCHIVE LIKE ORDERS;

# 数据归档存储过程
CREATE OR REPLACE PROCEDURE SP_ARCHIVE_ORDERS(
IN p_archive_date DATE
)
LANGUAGE SQL
BEGIN
DECLARE v_count INTEGER;

— 统计需要归档的数据量
SELECT COUNT(*) INTO v_count
FROM ORDERS
WHERE CREATE_TIME < p_archive_date; -- 插入归档表 INSERT INTO ORDERS_ARCHIVE SELECT * FROM ORDERS WHERE CREATE_TIME < p_archive_date; -- 删除原表数据 DELETE FROM ORDERS WHERE CREATE_TIME < p_archive_date; -- 记录归档日志 INSERT INTO ARCHIVE_LOG ( TABLE_NAME, ARCHIVE_DATE, ARCHIVE_COUNT, ARCHIVE_TIME ) VALUES ( 'ORDERS', p_archive_date, v_count, CURRENT TIMESTAMP ); COMMIT; END; # 执行归档 CALL SP_ARCHIVE_ORDERS(CURRENT DATE - 365 DAYS);

# 创建数据清理规则表
CREATE TABLE DATA_CLEANUP_RULES (
RULE_ID INTEGER NOT NULL GENERATED ALWAYS AS IDENTITY,
TABLE_NAME VARCHAR(100) NOT NULL,
CONDITION_SQL VARCHAR(500) NOT NULL,
RETENTION_DAYS INTEGER NOT NULL,
IS_ACTIVE CHAR(1) DEFAULT ‘Y’,
CREATE_TIME TIMESTAMP NOT NULL DEFAULT CURRENT TIMESTAMP,
CONSTRAINT PK_DATA_CLEANUP_RULES PRIMARY KEY (RULE_ID)
);

INSERT INTO DATA_CLEANUP_RULES (TABLE_NAME, CONDITION_SQL, RETENTION_DAYS)
VALUES
(‘LOGS’, ‘LOG_TIME < CURRENT DATE - RETENTION_DAYS DAYS', 30), ('TEMP_DATA', 'CREATE_TIME < CURRENT DATE - RETENTION_DAYS DAYS', 7), ('AUDIT_LOG', 'AUDIT_TIME < CURRENT DATE - RETENTION_DAYS DAYS', 365); # 数据清理存储过程 CREATE OR REPLACE PROCEDURE SP_CLEANUP_DATA() LANGUAGE SQL BEGIN DECLARE v_table_name VARCHAR(100); DECLARE v_condition_sql VARCHAR(500); DECLARE v_retention_days INTEGER; DECLARE v_delete_count INTEGER; FOR cleanup_cursor AS SELECT TABLE_NAME, CONDITION_SQL, RETENTION_DAYS FROM DATA_CLEANUP_RULES WHERE IS_ACTIVE = 'Y' DO SET v_table_name = cleanup_cursor.TABLE_NAME; SET v_condition_sql = cleanup_cursor.CONDITION_SQL; SET v_retention_days = cleanup_cursor.RETENTION_DAYS; -- 替换变量 SET v_condition_sql = REPLACE(v_condition_sql, 'RETENTION_DAYS', v_retention_days); -- 执行删除 SET v_delete_count = 0; -- 记录清理日志 INSERT INTO CLEANUP_LOG ( TABLE_NAME, DELETE_COUNT, CLEANUP_TIME ) VALUES ( v_table_name, v_delete_count, CURRENT TIMESTAMP ); END FOR; END;