大数据领域数据预处理：保障数据安全与隐私

    import re
    import hashlib
    import random
    import string
    
    class DataMasking:
    @staticmethod
    def replace_masking(text, keep_last=4, mask_char='*'):
        """保留最后几位，其余替换为指定字符"""
        if len(text) <= keep_last:
            return text
        return mask_char * (len(text)-keep_last) + text[-keep_last:]
    
    @staticmethod
    def hash_masking(text, salt=None, algorithm='sha256'):
        """哈希脱敏，不可逆"""
        if salt is None:
            salt = ''.join(random.choices(string.ascii_letters + string.digits, k=8))
        hash_obj = hashlib.new(algorithm)
        hash_obj.update((text + salt).encode('utf-8'))
        return hash_obj.hexdigest()
    
    @staticmethod
    def email_masking(email):
        """邮箱脱敏处理"""
        parts = email.split('@')
        if len(parts) != 2:
            return email
        username = parts[0]
        domain = parts[1]
        if len(username) <= 1:
            masked_username = '*'
        else:
            masked_username = username[0] + '****' + (username[-1] if len(username) > 1 else '')
        return f"{masked_username}@{domain}"
    
    @staticmethod
    def shuffle_masking(text):
        """随机打乱顺序"""
        chars = list(text)
        random.shuffle(chars)
        return ''.join(chars)
    
    
    python
    
    
![](https://ad.itadn.com/c/weblog/blog-img/images/2025-08-16/A3VeUZM8istoClmDQjG2r5L46nHE.png)

    import numpy as np
    
    class DifferentialPrivacy:
    @staticmethod
    def laplace_mechanism(true_answer, sensitivity, epsilon):
        """
        拉普拉斯机制实现差分隐私
        :param true_answer: 真实查询结果
        :param sensitivity: 查询的敏感度
        :param epsilon: 隐私预算
        :return: 满足差分隐私的噪声结果
        """
        scale = sensitivity / epsilon
        noise = np.random.laplace(0, scale)
        return true_answer + noise
    
    @staticmethod
    def exponential_mechanism(utility_scores, sensitivity, epsilon):
        """
        指数机制实现差分隐私
        :param utility_scores: 各选项的效用分数字典 {option: score}
        :param sensitivity: 效用函数的敏感度
        :param epsilon: 隐私预算
        :return: 根据差分隐私选择的选项
        """
        # 计算每个选项的选择概率
        probabilities = {}
        max_score = max(utility_scores.values())
        for option, score in utility_scores.items():
            probabilities[option] = np.exp(epsilon * score / (2 * sensitivity))
        
        # 归一化概率
        total = sum(probabilities.values())
        normalized_prob = {k: v/total for k, v in probabilities.items()}
        
        # 根据概率随机选择
        options = list(normalized_prob.keys())
        probs = list(normalized_prob.values())
        return np.random.choice(options, p=probs)
    
    
    python
    
    
![](https://ad.itadn.com/c/weblog/blog-img/images/2025-08-16/nMWola9RJkt4hUqVGv1zcgZBNQ3m.png)

    from Pyfhel import Pyfhel, PyPtxt, PyCtxt
    
    class HomomorphicEncryption:
    def __init__(self):
        self.HE = Pyfhel()
        # 参数设置
        self.params = {
            'scheme': 'BFV',  # 也可以选择CKKS
            'n': 2**13,       # 多项式模数
            't_bits': 20,     # 明文模数的位数
            'sec': 128,       # 安全级别
        }
    
    def generate_keys(self):
        """生成公私钥对"""
        self.HE.contextGen(**self.params)
        self.HE.keyGen()
        self.HE.relinKeyGen()
        self.HE.rotateKeyGen()
    
    def encrypt(self, plaintext):
        """加密数据"""
        return self.HE.encrypt(plaintext)
    
    def decrypt(self, ciphertext):
        """解密数据"""
        return self.HE.decrypt(ciphertext)
    
    def homomorphic_add(self, ctxt1, ctxt2):
        """同态加法"""
        return ctxt1 + ctxt2
    
    def homomorphic_multiply(self, ctxt1, ctxt2):
        """同态乘法"""
        return ctxt1 * ctxt2
    
    def homomorphic_dot_product(self, encrypted_vec1, encrypted_vec2):
        """加密向量的点积"""
        if len(encrypted_vec1) != len(encrypted_vec2):
            raise ValueError("Vectors must be of same length")
        
        result = self.encrypt(0)
        for v1, v2 in zip(encrypted_vec1, encrypted_vec2):
            product = self.homomorphic_multiply(v1, v2)
            result = self.homomorphic_add(result, product)
        return result
    
    
    python
    
    
![](https://ad.itadn.com/c/weblog/blog-img/images/2025-08-16/PZn5WYSO0tw3IfHueGg49hEqRxVJ.png)

    pip install numpy pandas Pyfhel
    
    
    bash

    import pandas as pd
    import numpy as np
    from datetime import datetime
    from typing import List, Dict, Any
    
    class PrivacyPreservingPreprocessor:
    def __init__(self, epsilon: float = 1.0, k: int = 3):
        """
        初始化隐私保护预处理器
        :param epsilon: 差分隐私的隐私预算
        :param k: k-匿名中的k值
        """
        self.epsilon = epsilon
        self.k = k
        self.qid_columns = None  # 准标识符列
        self.sensitive_columns = None  # 敏感列
        self.column_types = None  # 列类型
        
    def detect_column_types(self, df: pd.DataFrame) -> Dict[str, str]:
        """自动检测列的数据类型"""
        column_types = {}
        for col in df.columns:
            # 简单类型检测，实际应用中需要更复杂的逻辑
            if df[col].dtype == 'object':
                if df[col].str.contains('@').any():
                    column_types[col] = 'email'
                elif df[col].str.match(r'^\d{3}-\d{2}-\d{4}$').any():
                    column_types[col] = 'ssn'
                elif df[col].str.match(r'^\d{10}$').any():
                    column_types[col] = 'phone'
                else:
                    column_types[col] = 'categorical'
            elif np.issubdtype(df[col].dtype, np.number):
                if len(df[col].unique()) < 20:
                    column_types[col] = 'categorical'
                else:
                    column_types[col] = 'numerical'
            elif np.issubdtype(df[col].dtype, np.datetime64):
                column_types[col] = 'datetime'
        return column_types
    
    def apply_k_anonymity(self, df: pd.DataFrame, qid_columns: List[str]) -> pd.DataFrame:
        """
        应用k-匿名化
        :param df: 输入数据框
        :param qid_columns: 准标识符列
        :return: 满足k-匿名的数据框
        """
        self.qid_columns = qid_columns
        
        # 1. 泛化处理
        generalized_df = df.copy()
        for col in qid_columns:
            if self.column_types[col] == 'numerical':
                # 数值型数据使用区间泛化
                bins = self._calculate_optimal_bins(df[col], self.k)
                generalized_df[col] = pd.cut(df[col], bins=bins, include_lowest=True)
            elif self.column_types[col] == 'datetime':
                # 日期数据按年或月泛化
                generalized_df[col] = generalized_df[col].dt.to_period('Y')
            elif self.column_types[col] in ['categorical', 'email', 'phone', 'ssn']:
                # 类别数据使用更一般的类别
                generalized_df[col] = generalized_df[col].apply(self._generalize_categorical)
        
        # 2. 检查并处理不满足k-匿名的组
        group_counts = generalized_df.groupby(qid_columns).size()
        small_groups = group_counts[group_counts < self.k].index
        
        if not small_groups.empty:
            # 处理小群体的策略：进一步泛化或抑制
            for group in small_groups:
                mask = True
                for col, val in zip(qid_columns, group):
                    mask &= (generalized_df[col] == val)
                indices = generalized_df[mask].index
                
                # 简单策略：删除不满足k-匿名的记录
                generalized_df = generalized_df.drop(indices)
        
        return generalized_df
    
    def _calculate_optimal_bins(self, series: pd.Series, k: int) -> List[float]:
        """计算最优的分箱边界，确保每个箱至少有k个样本"""
        sorted_values = np.sort(series.dropna().unique())
        bins = [sorted_values[0] - 1]  # 起始边界
        
        i = 0
        n = len(sorted_values)
        while i < n:
            j = min(i + k, n)
            if j < n:
                # 确保边界在两个不同值之间
                boundary = (sorted_values[j-1] + sorted_values[j]) / 2
                bins.append(boundary)
            i = j
        
        bins.append(sorted_values[-1] + 1)  # 结束边界
        return bins
    
    def _generalize_categorical(self, value: Any) -> str:
        """泛化类别数据"""
        if pd.isna(value):
            return 'NA'
        
        if self.column_types.get('email', None) == 'email':
            return DataMasking.email_masking(value)
        elif self.column_types.get('phone', None) == 'phone':
            return DataMasking.replace_masking(str(value), keep_last=2)
        elif self.column_types.get('ssn', None) == 'ssn':
            return DataMasking.replace_masking(str(value), keep_last=4)
        else:
            # 简单首字母泛化
            return str(value)[0] + '***'
    
    def apply_differential_privacy(self, df: pd.DataFrame, numerical_columns: List[str]) -> pd.DataFrame:
        """
        对数值列应用差分隐私
        :param df: 输入数据框
        :param numerical_columns: 需要应用差分隐私的数值列
        :return: 添加噪声后的数据框
        """
        dp_df = df.copy()
        for col in numerical_columns:
            if col not in df.columns:
                continue
                
            # 计算敏感度(这里使用全局敏感度)
            sensitivity = df[col].max() - df[col].min()
            
            # 应用拉普拉斯机制
            dp_df[col] = df[col].apply(
                lambda x: DifferentialPrivacy.laplace_mechanism(x, sensitivity, self.epsilon)
            )
        return dp_df
    
    def apply_data_masking(self, df: pd.DataFrame, sensitive_columns: List[str]) -> pd.DataFrame:
        """
        应用数据脱敏
        :param df: 输入数据框
        :param sensitive_columns: 敏感列
        :return: 脱敏后的数据框
        """
        self.sensitive_columns = sensitive_columns
        masked_df = df.copy()
        
        for col in sensitive_columns:
            if col not in df.columns:
                continue
                
            if self.column_types[col] == 'email':
                masked_df[col] = df[col].apply(DataMasking.email_masking)
            elif self.column_types[col] == 'phone':
                masked_df[col] = df[col].apply(lambda x: DataMasking.replace_masking(str(x), keep_last=2))
            elif self.column_types[col] == 'ssn':
                masked_df[col] = df[col].apply(lambda x: DataMasking.replace_masking(str(x), keep_last=4))
            elif self.column_types[col] == 'categorical':
                masked_df[col] = df[col].apply(lambda x: DataMasking.hash_masking(str(x)))
        
        return masked_df
    
    def preprocess_pipeline(self, df: pd.DataFrame, 
                          qid_columns: List[str], 
                          sensitive_columns: List[str],
                          numerical_columns: List[str]) -> pd.DataFrame:
        """
        完整的隐私保护预处理管道
        """
        # 1. 检测列类型
        self.column_types = self.detect_column_types(df)
        
        # 2. 应用k-匿名
        anonymized_df = self.apply_k_anonymity(df, qid_columns)
        
        # 3. 应用差分隐私
        dp_df = self.apply_differential_privacy(anonymized_df, numerical_columns)
        
        # 4. 应用数据脱敏
        final_df = self.apply_data_masking(dp_df, sensitive_columns)
        
        return final_df
    
    
    python
    
    
![](https://ad.itadn.com/c/weblog/blog-img/images/2025-08-16/MsKUEA3mXpLNvqaDlxuV2f8TOQbo.png)