数据脱敏实战指南：从原理到MySQL实现

CREATETABLE customers_masked AS
SELECT
    customer_id,
    CASE
        WHEN gender = 'M'THEN
            (SELECTCONCAT(
                ELT(1+FLOOR(RAND()*10), '张', '李', '王', '刘', '陈'),
                ELT(1+FLOOR(RAND()*10), '伟', '强', '军', '明', '勇'


    
)
            ))
        ELSE
            (SELECTCONCAT(
                ELT(1+FLOOR(RAND()*10), '张', '李', '王', '刘', '陈'),
                ELT(1+FLOOR(RAND()*10), '芳', '娜', '静', '敏', '婷')
            ))
    ENDAS masked_name
FROM customers;

SELECT
    customer_id,
    CONCAT(
        LEFT(credit_card_number, 4),
        REPEAT('*', LENGTH(credit_card_number) - 8),
        RIGHT(credit_card_number, 4)
    ) AS masked_card_number
FROM payment_info;

--实际输出效果: '4356********1234'

UPDATE employees_test SET
    salary = ROUND(
        -- 在原始薪资的±15%范围内随机化
        salary * (0.85 + (RAND() * 0.3)),
        2
    );

CREATETABLE shuffled_purchases AS
SELECT
    p.purchase_id,
    (SELECT customer_id FROM purchases ORDERBYRAND() LIMIT1) AS shuffled_customer_id,
    p.product_id,
    p.purchase_date,
    p.amount
FROM purchases p;

-- 创建加密函数
CREATEFUNCTION encrypt_data(p_data VARCHAR(255), p_key VARCHAR(32))
RETURNSVARCHAR(255)
BEGIN
    RETURNTO_BASE64(AES_ENCRYPT(p_data, p_key));
END;

-- 加密存储病历信息
INSERTINTO patient_records (
    patient_id, 
    encrypted_diagnosis
) VALUES (
    101, 
    encrypt_data('高血压，需定期监测', '安全密钥需妥善保管')
);

-- 创建银行数据库
CREATEDATABASE bank_system;
USE bank_system;

-- 创建客户信息表
CREATETABLE customers (
    customer_id INT AUTO_INCREMENT PRIMARY KEY,
    nameVARCHAR(50) NOTNULL,
    id_number VARCHAR(18) NOTNULL,
    phone VARCHAR(15) NOTNULL,
    email VARCHAR(


    
50),
    address VARCHAR(100),
    birth_date DATE,
    registration_date DATETIME DEFAULTCURRENT_TIMESTAMP
);

-- 创建账户信息表
CREATETABLE accounts (
    account_id INT AUTO_INCREMENT PRIMARY KEY,
    customer_id INTNOTNULL,
    account_number VARCHAR(20) NOTNULL,
    account_type VARCHAR(20) NOTNULL,
    balance DECIMAL(15,2) NOTNULL,
    open_date DATENOTNULL,
    statusVARCHAR(10) DEFAULT'active',
    FOREIGN KEY (customer_id) REFERENCES customers(customer_id)
);




    
-- 创建交易记录表
CREATETABLE transactions (
    transaction_id INT AUTO_INCREMENT PRIMARY KEY,
    account_id INTNOTNULL,
    transaction_date DATETIME NOTNULL,
    amount DECIMAL(15,2) NOTNULL,
    transaction_type VARCHAR(20) NOTNULL,
    description TEXT,
    FOREIGN KEY (account_id) REFERENCES accounts(account_id)
);

-- 创建开发环境数据库
CREATEDATABASE bank_system_dev;
USE bank_system_dev;

-- 创建脱敏函数



    
CREATEFUNCTION mask_id_number(idVARCHAR(18)) 
RETURNSVARCHAR(18)
BEGIN
    RETURNCONCAT(LEFT(id, 1), REPEAT('*', LENGTH(id) - 4), RIGHT(id, 3));
END;

CREATEFUNCTION mask_phone(phone VARCHAR(15)) 
RETURNSVARCHAR(15)
BEGIN
    RETURNCONCAT(LEFT(phone, 3), REPEAT('*', LENGTH(phone) - 7), 


    
RIGHT(phone, 4));
END;

CREATEFUNCTION mask_email(email VARCHAR(50)) 
RETURNSVARCHAR(50)
BEGIN
    DECLARE at_pos INT;
    SET at_pos = POSITION('@'IN email);
    RETURN CONCAT('user', RIGHT(email, LENGTH(email) - at_pos + 1));
END;

-- 插入脱敏后的数据
INSERTINTO bank_system_dev.customers
SELECT
    customer_id,
    CONCAT('用户', customer_id), -- 替换姓名
    mask_id_number(id_number),
    mask_phone(phone),
    mask_email(email),
    mask_address(address),
    birth_date,
    registration_date
FROM bank_system.customers;

-- 创建客户信息脱敏视图
CREATEVIEW bank_system.v_customers_masked AS



    
SELECT
    customer_id,
    CONCAT('客户', LPAD(customer_id, 4, '0')) ASname,
    CONCAT(LEFT(id_number, 2), REPEAT('*', LENGTH(id_number) - 4), RIGHT(id_number, 2)) AS id_number,
    CONCAT(LEFT(phone, 3), REPEAT('*', LENGTH(phone) - 5), RIGHT(phone, 2)) AS phone,
    CONCAT(LEFT(SUBSTRING_INDEX(email, '@', 1), 1), '***@', SUBSTRING_INDEX(email, '@', -1)) AS email
FROM bank_system.customers;




    
-- 创建账户信息脱敏视图
CREATEVIEW bank_system.v_accounts_masked AS
SELECT
    account_id,
    customer_id,
    CONCAT('****', RIGHT(account_number, 4)) AS account_number,
    account_type,
    CASE
        WHEN balance 0THEN-1
        WHEN balance BETWEEN0AND10000THEN1
        WHEN balance BETWEEN10001AND50000THEN2
        WHEN balance BETWEEN50001AND100000THEN3
        ELSE4
    END


    
AS balance_level,
    YEAR(open_date) AS open_year,
    status
FROM bank_system.accounts;

-- 为不同分析团队创建过滤视图
-- 零售银行分析团队 - 只能看到个人客户
CREATEVIEW bank_system.v_retail_analysis AS
SELECT c.*, a.account_type, a.balance_level
FROM bank_system.v_customers_masked c
JOIN bank_system.v_accounts_masked a ON c.customer_id = a.customer_id
WHERE a.account_type IN ('savings', 'checking', 'personal');

-- 企业银行分析团队 - 只能看到企业客户
CREATEVIEW bank_system.v_corporate_analysis AS
SELECT c.*, a.account_type, a.balance_level
FROM bank_system.v_customers_masked c
JOIN bank_system.v_accounts_masked a ON c.customer_id = a.customer_id
WHERE a.account_type IN ('corporate', 'business');

-- 检查开发环境脱敏数据
SELECT * FROM bank_system_dev.customers LIMIT5;
SELECT * FROM bank_system_dev.accounts LIMIT5;

-- 检查分析视图脱敏效果
SELECT * FROM bank_system.v_customers_masked LIMIT5;
SELECT * FROM bank_system.v_retail_analysis LIMIT5;

    
-- 测试组合查询是否能识别特定个人
SELECT * 
FROM bank_system.v_customers_masked 
WHERE birth_year = '1980-**-**'
ANDLEFT(address, 9) = '北京市海淀区';

-- 创建定期刷新开发环境数据的存储过程
CREATEPROCEDURE refresh_dev_data()
BEGIN
    -- 清空现有数据
    TRUNCATETABLE bank_system_dev.transactions;
    TRUNCATETABLE


    
 bank_system_dev.accounts;
    TRUNCATETABLE bank_system_dev.customers;
    
    -- 插入脱敏后的新数据
    -- (重复前面的插入脱敏数据的SQL)
    
    -- 记录执行日志
    INSERTINTO data_refresh_log (refresh_time, status, description)
    VALUES (NOW(), 'SUCCESS', 'Development database refreshed with masked data');
END;

-- 创建事件定期执行
CREATEEVENT refresh_dev_data_weekly
ON SCHEDULE EVERY 1WEEK
DOCALL refresh_dev_data();

-- 创建通用脱敏函数库
CREATESCHEMA masking_lib;

-- 在公共函数库中创建手机号掩码函数
CREATEFUNCTION masking_lib.mask_phone(phone VARCHAR(


    
20)) 
RETURNSVARCHAR(20)
BEGIN
RETURNCONCAT(LEFT(phone, 3), '****', RIGHT(phone, 4));
END;

-- 在CRM系统中使用
CREATEVIEW crm.customer_view AS
SELECT customer_id, name, masking_lib.mask_phone(phone) AS phone
FROM crm.customers;

-- 在订单系统中使用
CREATEVIEW orders.order_contact_view AS
SELECT order_id, customer_id, masking_lib.mask_phone(contact_phone) AS contact_phone
FROM orders.order_details;

-- 创建映射表
CREATETABLE id_mapping (
  original_id VARCHAR(50),



    
  masked_id VARCHAR(50),
  mapping_type VARCHAR(20),
  PRIMARY KEY (original_id, mapping_type)
);

-- 创建映射函数
CREATEFUNCTION get_masked_id(original_id VARCHAR(50), mapping_type VARCHAR(20))
RETURNSVARCHAR(50)
BEGIN
DECLARE masked_id VARCHAR(50);

-- 检查映射是否已存在
SELECT m.masked_id INTO masked_id
FROM id_mapping m
WHERE m.original_id = original_id AND m.mapping_type = mapping_type;

-- 如果不存在，创建新映射
  IF masked_id IS NULL THEN
    SET masked_id = CONCAT('MASK', LPAD(CONV(FLOOR(RAND() * 16777215), 10, 16), 


    
6, '0'));
    
    INSERTINTO id_mapping (original_id, masked_id, mapping_type)
    VALUES (original_id, masked_id, mapping_type);
ENDIF;

  RETURN masked_id;
END;

-- 医生视图（最低脱敏级别）
CREATEVIEW patient_data_for_doctors AS
SELECT
  patient_id,
name,
  birth_date,
-- 完整地址对医生可见
  address,
-- 完整联系方式对医生可见
  phone,
-- 完整医疗记录对医生可见
  medical_history
FROM patients;

-- 护士视图（中度脱敏）
CREATEVIEW patient_data_for_nurses AS
SELECT
  patient_id,
name,
YEAR(birth_date) AS birth_year,
-- 地址部分可见
CONCAT(LEFT(address, LOCATE(',', address) - 


    
1), ', ***') AS partial_address,
-- 电话部分可见
CONCAT(LEFT(phone, 3), '****', RIGHT(phone, 4)) AS masked_phone,
-- 医疗记录摘要
LEFT(medical_history, 100) AS medical_summary
FROM patients;

CREATEPROCEDURE mask_data(
IN p_mask_level ENUM('LOW', 'MEDIUM', 'HIGH'),
IN p_show_first INT,
IN p_show_last INT
)
BEGIN
DECLARE v_mask_char CHAR(1);

-- 根据脱敏级别设置掩码字符
  CASE p_mask_level
    WHEN 'LOW' THEN 


    
SET v_mask_char = 'x';
    WHEN 'MEDIUM' THEN SET v_mask_char = '*';
    WHEN 'HIGH' THEN SET v_mask_char = '#';
ENDCASE;

-- 创建动态视图
SET @sql = CONCAT('
    CREATE OR REPLACE VIEW masked_customer_contacts AS
    SELECT
      customer_id,
      name,
      CASE 
        WHEN LENGTH(phone) <= ', p_show_first + p_show_last, ' THEN phone
        ELSE CONCAT(
          LEFT(phone, ', p_show_first, '),
          REPEAT("', v_mask_char, '", LENGTH(phone) - ', p_show_first + p_show_last, '),
          RIGHT(phone, ', p_show_last, ')
        )
      END AS masked_phone
    FROM customers
  ');

PREPARE stmt FROM @sql;
EXECUTE stmt;
DEALLOCATEPREPARE stmt;
END;

-- 使用示例 - 低级别脱敏，显示前3位和后4位
CALL mask_data('LOW', 3, 4);

    
-- 创建脱敏物化视图表
CREATETABLE masked_transactions_mv (
  transaction_id INT PRIMARY KEY,
  masked_account VARCHAR(20),
  transaction_date DATE,
  amount_category VARCHAR(20),
  last_refresh TIMESTAMP
);

-- 使用增量更新方式刷新
INSERTINTO masked_transactions_mv
SELECT
  t.transaction_id,
CONCAT('****', RIGHT(t.account_number, 4)) AS masked_account,
  t.transaction_date,
CASE
    WHEN t.amount 100THEN'small'
    WHEN t.amount BETWEEN100AND1000THEN'medium'
    ELSE'large'
ENDAS amount_category,



    
NOW() AS last_refresh
FROM transactions t
WHERENOTEXISTS (
SELECT1FROM masked_transactions_mv mv
WHERE mv.transaction_id = t.transaction_id
);

-- 问题：金融分析团队反馈交易金额完全随机化后无法进行趋势分析
-- 解决方案：使用保持数据分布特性的脱敏方法

CREATEFUNCTION distribution_preserving_amount_mask(
  original_amount DECIMAL(15,2),
  segment_id INT-- 可用于保持按客户细分的金额分布
)
RETURNSDECIMAL(15,2


    
)
BEGIN
DECLARE avg_amount DECIMAL(15,2);
DECLARE std_dev DECIMAL(15,2);
DECLARE masked_amount DECIMAL(15,2);

-- 获取该细分市场的平均交易金额和标准差
SELECTAVG(amount), STDDEV(amount) INTO avg_amount, std_dev
FROM transactions
WHERE customer_segment = segment_id;

-- 计算原始金额的Z分数（标准分数）
SET @z_score = (original_amount - avg_amount) / std_dev;

-- 使用同样的Z分数生成新金额，但添加小的随机噪声
SET masked_amount = avg_amount + (@z_score * std_dev * (0.97 + (RAND() * 0.06)));

-- 四舍五入到合理精度
  RETURN ROUND(masked_amount, 2);
END;

-- 问题：客户服务系统中动态脱敏查询响应时间超过2秒
-- 解决方案：创建脱敏物化视图并添加索引

CREATETABLE customer_service.masked_customer_data (
  customer_id INT PRIMARY KEY,
  masked_name VARCHAR(100),
  masked_email VARCHAR(100),
  masked_phone VARCHAR(20),
  account_status VARCHAR(20),
  last_contact_date DATE,
  last_refresh_time TIMESTAMP,
INDEX idx_phone (masked_phone),
INDEX idx_email (masked_email)
);

-- 优化后查询(耗时<50ms):
SELECT
  customer_id,
  masked_name,
  masked_email,
  masked_phone,
  account_status
FROM customer_service.masked_customer_data
WHERE masked_phone = '135****1234';

-- 问题：销售部门和财务部门使用不同脱敏方法导致客户ID无法关联
-- 解决方案：建立统一的客户ID映射表

CREATETABLE data_governance.customer_id_mapping (
  original_id VARCHAR(50) PRIMARY KEY,
  sales_masked_id VARCHAR(50) NOTNULL,
  finance_masked_id VARCHAR(50) NOTNULL,
  marketing_masked_id VARCHAR(50) NOTNULL,
  created_at TIMESTAMPDEFAULTCURRENT_TIMESTAMP,
UNIQUEKEY idx_sales (sales_masked_id),
UNIQUEKEY idx_finance (finance_masked_id),
UNIQUEKEY idx_marketing (marketing_masked_id)
);

-- 创建跨部门分析视图



    
CREATEVIEW cross_department.customer_analysis AS
SELECT
  s.masked_customer_id AS sales_id,
  f.masked_customer_id AS finance_id,
  s.region,
  s.customer_type,
  f.account_status,
  f.credit_rating
FROM sales.customers_masked s
JOIN data_governance.customer_id_mapping m 
ON s.masked_customer_id = m.sales_masked_id
JOIN finance.customers_masked f 
ON m.finance_masked_id = f.masked_customer_id;

-- 问题：研究团队发现年龄+邮编+性别组合可能导致唯一标识个人
-- 解决方案：实施K匿名性(K=5)脱敏

CREATE FUNCTIONcheck_k_anonymity(
  p_table_name VARCHAR(100),
  p_column_list VARCHAR(255), -- 逗号分隔的列名
  p_min_k INT -- 最小K值，通常>=


    
5
)
RETURNSBOOLEAN
BEGIN
  DECLAREv_countINT;
DECLARE v_violations INT;

  -- 动态构建SQL统计不满足K匿名性的组合数
  SET @sql = CONCAT('
    SELECT COUNT(*) INTO @violations
    FROM (
      SELECT COUNT(*) as group_size
      FROM ', p_table_name, '
      GROUP BY ', p_column_list, '
      HAVING group_size < ', p_min_k, '
    ) AS violation_groups
  ');

  PREPARE stmt FROM @sql;
  EXECUTE stmt;
  DEALLOCATE PREPARE stmt;

  -- 如果存在违反K匿名性的组合，返回FALSE
IF @violations > 0 THEN
    RETURNFALSE;
ELSE
    RETURNTRUE;
  END IF;
END;