本文目录导读：
我来介绍使用Pandas进行数据哈希编码的几种常用方法：
基本哈希编码
import pandas as pd
import hashlib
# 创建示例数据
df = pd.DataFrame({
'user_id': ['user001', 'user002', 'user003', 'user001', 'user004'],
'name': ['张三', '李四', '王五', '张三', '赵六'],
'city': ['北京', '上海', '广州', '北京', '深圳']
})
# 使用hashlib进行MD5编码
def md5_encode(value):
if pd.isna(value):
return None
return hashlib.md5(str(value).encode()).hexdigest()[:8] # 取前8位
# 对特定列进行哈希编码
df['user_id_hash'] = df['user_id'].apply(md5_encode)
df['name_hash'] = df['name'].apply(md5_encode)
print("原始数据及哈希编码：")
print(df)
使用Pandas内置哈希函数
# 使用pandas的hash函数
import pandas as pd
# 创建更复杂的数据
df = pd.DataFrame({
'user_id': ['U001', 'U002', 'U003', 'U001', 'U004'],
'email': ['a@qq.com', 'b@qq.com', 'c@qq.com', 'a@qq.com', 'd@qq.com'],
'age': [25, 30, 28, 25, 35]
})
# 使用pandas的factorize进行编码（类似哈希）
def hash_encode(series, method='hash'):
"""多种哈希编码方法"""
if method == 'factorize':
# 使用factorize（适合分类数据）
codes, uniques = pd.factorize(series)
return pd.Series(codes, index=series.index)
elif method == 'hash':
# 使用hash函数
return series.apply(lambda x: hash(str(x)) % 10000)
elif method == 'categorical':
# 转换为分类类型编码
return pd.Categorical(series).codes
# 应用不同的哈希方法
df['user_id_factorize'] = hash_encode(df['user_id'], 'factorize')
df['user_id_hash'] = hash_encode(df['user_id'], 'hash')
df['email_categorical'] = hash_encode(df['email'], 'categorical')
print("\n不同哈希方法对比：")
print(df)
批量哈希编码
import pandas as pd
import hashlib
class DataHasher:
"""数据哈希编码器"""
def __init__(self, method='sha256', length=8):
self.method = method
self.length = length
self.mapping = {}
def hash_value(self, value):
"""编码单个值"""
if pd.isna(value):
return None
# 使用映射确保一致性
if value not in self.mapping:
hash_func = getattr(hashlib, self.method)
hash_result = hash_func(str(value).encode()).hexdigest()
self.mapping[value] = hash_result[:self.length]
return self.mapping[value]
def encode_column(self, df, column_name, new_column_name=None):
"""编码整列数据"""
if new_column_name is None:
new_column_name = f"{column_name}_hash"
df[new_column_name] = df[column_name].apply(self.hash_value)
return df
def encode_columns(self, df, columns, suffix='_hash'):
"""编码多列数据"""
for col in columns:
new_col = f"{col}{suffix}"
df[new_col] = df[col].apply(self.hash_value)
return df
# 使用示例
data = pd.DataFrame({
'user_id': ['U001', 'U002', 'U003', 'U001', 'U004'],
'email': ['a@test.com', 'b@test.com', 'c@test.com', 'a@test.com', 'd@test.com'],
'phone': ['13800138001', '13900139002', '13700137003', '13800138001', '13600136004']
})
# 创建哈希编码器
hasher = DataHasher(method='sha256', length=6)
# 编码单列
df_encoded = hasher.encode_column(data.copy(), 'user_id')
# 编码多列
df_encoded = hasher.encode_columns(df_encoded, ['email', 'phone'])
print("\n批量哈希编码结果：")
print(df_encoded)
# 验证编码一致性
print("\n编码一致性验证：")
print(f"'U001' 的哈希值: {hasher.hash_value('U001')}")
print(f"'a@test.com' 的哈希值: {hasher.hash_value('a@test.com')}")
实际应用：数据脱敏
import pandas as pd
import hashlib
import numpy as np
# 创建敏感数据
df_sensitive = pd.DataFrame({
'name': ['张三', '李四', '王五', '张三'],
'id_card': ['110101199001011234', '110101199002022345', '110101199003033456', '110101199001011234'],
'phone': ['13812345678', '13923456789', '13734567890', '13812345678'],
'salary': [15000, 20000, 18000, 15000]
})
def data_masking(df, sensitive_columns, hash_length=8):
"""数据脱敏函数"""
masked_df = df.copy()
for col in sensitive_columns:
if col in df.columns:
# 对敏感列进行哈希编码
masked_df[f'{col}_masked'] = df[col].apply(
lambda x: hashlib.md5(str(x).encode()).hexdigest()[:hash_length]
)
# 保留原始数据的前后几位
masked_df[f'{col}_partial'] = df[col].apply(
lambda x: str(x)[:3] + '*' * (len(str(x))-6) + str(x)[-3:] if len(str(x)) > 6 else x
)
# 删除原始敏感列（可选）
# masked_df = masked_df.drop(columns=sensitive_columns)
return masked_df
# 执行数据脱敏
df_masked = data_masking(df_sensitive, ['id_card', 'phone'])
print("原始数据：")
print(df_sensitive)
print("\n脱敏后数据：")
print(df_masked)
哈希编码性能对比
import pandas as pd
import hashlib
import time
# 创建大数据集
n = 100000
df_large = pd.DataFrame({
'value': [f'data_{i}' for i in range(n)] + ['重复数据'] * 1000
})
def benchmark_hash_methods(df, column):
"""性能测试"""
results = {}
# 方法1: 使用hashlib
start = time.time()
df['hash_md5'] = df[column].apply(lambda x: hashlib.md5(str(x).encode()).hexdigest()[:8])
results['hashlib_md5'] = time.time() - start
# 方法2: 使用pandas hash
start = time.time()
df['hash_pd'] = df[column].apply(lambda x: hash(str(x)) % 100000)
results['pandas_hash'] = time.time() - start
# 方法3: 使用factorize
start = time.time()
codes, _ = pd.factorize(df[column])
df['hash_factorize'] = codes
results['factorize'] = time.time() - start
# 方法4: 使用lambda和map
start = time.time()
unique_values = df[column].unique()
hash_map = {val: hashlib.md5(str(val).encode()).hexdigest()[:8] for val in unique_values}
df['hash_map'] = df[column].map(hash_map)
results['mapping'] = time.time() - start
return results
# 执行性能测试
perf_results = benchmark_hash_methods(df_large, 'value')
print("\n哈希编码性能对比（100k条数据）：")
for method, time_taken in sorted(perf_results.items(), key=lambda x: x[1]):
print(f"{method:20s}: {time_taken:.4f} 秒")
实用工具函数
def simple_hash_encode(df, columns, method='md5', prefix='hash_'):
"""
简单实用的哈希编码函数
参数:
df: DataFrame
columns: 需要编码的列名列表
method: 哈希方法 ('md5', 'sha256', 'pandas_hash')
prefix: 新列名前缀
返回:
DataFrame with encoded columns
"""
result = df.copy()
for col in columns:
if col not in df.columns:
continue
if method == 'pandas_hash':
# 使用pandas的hash方法
codes, _ = pd.factorize(df[col])
result[f'{prefix}{col}'] = codes
else:
# 使用hashlib
hash_func = hashlib.md5 if method == 'md5' else hashlib.sha256
result[f'{prefix}{col}'] = df[col].apply(
lambda x: hash_func(str(x).encode()).hexdigest()[:8]
)
return result
# 使用示例
df = pd.DataFrame({
'name': ['Alice', 'Bob', 'Charlie', 'Alice'],
'email': ['a@test.com', 'b@test.com', 'c@test.com', 'a@test.com']
})
df_encoded = simple_hash_encode(df, ['name', 'email'])
print("\n实用哈希编码结果：")
print(df_encoded)
注意事项
- 哈希碰撞：截取哈希值时要注意碰撞风险
- 一致性：相同输入必须产生相同输出
- 性能考量：大数据集建议使用映射或factorize
- 安全性：哈希编码不是加密，敏感数据仍需加密处理
这些方法可以根据实际需求选择使用,最常见的场景是数据脱敏和特征编码。
