MinHash算法性能优化时间2026年3月11日## MinHash算法介绍MinHashMinimum Hashing最小哈希是一种用于快速估算两个集合相似度的算法属于局部敏感哈希LSH技术的一种。### 核心原理MinHash算法的理论基础是Jaccard相似度。Jaccard相似度衡量两个集合A和B相似程度的指标计算方式为它们交集的大小与并集的大小之比。J(A, B) |A ∩ B| / |A ∪ B|其值在0到1之间值越接近1表示越相似。MinHash算法的巧妙之处MinHash发现了一个重要的概率关系——对于两个集合使用一个随机哈希函数对它们的所有元素进行哈希然后取各自最小的哈希值这两个最小哈希值相等的概率恰好等于它们的Jaccard相似度。P(h_min(A) h_min(B)) J(A, B)生成签名为了得到一个更稳定、更准确的估算MinHash会使用多个例如K个不同的哈希函数为每个集合生成K个最小哈希值这K个值就构成了该集合的“MinHash签名”。估算相似度最后通过比较两个集合的MinHash签名计算它们在K个位置上值相等的比例这个比例就是它们Jaccard相似度的近似值。### 算法实现1对于一段文字例如“high performance computing is interesting”通过n-grams我们将它分为[“high performance”“performance computing”“computing is”“is interesting”]在代码中对应变量为shingles2对于每个shingleMinHash算法通过Hash计算字符串对应的哈希值例如Hashshingle_1123452003计算出哈希值后算法通过长度为N的随机数组A和数组B对哈希值即12345200进行N维的随机打散生成signature指纹。这里N对应的即代码中的len_a和len_blen_a与len_b要求长度一致4之后算法会对这N维的signature指纹不断迭代取最小值min函数-- 这部分后面会通过SVE向量化指令进行加速5最终的结果及为Jaccard相似度Python简单实现直接通过datasketch中的MinHash计算from datasketch import MinHashdef compute_minhash_signature(text, num_perm128):“”计算单个文本的 MinHash 签名“”shingles text.lower().split()# 2. 初始化 MinHash 对象m MinHash(num_permnum_perm)# 3. 更新签名# datasketch 会自动处理哈希和取最小值的逻辑for s in shingles:m.update(s.encode(‘utf8’))return m# demodata1 “我爱吃苹果和香蕉”data2 “我爱吃苹果和橘子”# 计算签名sig1 compute_minhash_signature(data1)sig2 compute_minhash_signature(data2)# 估算 Jaccard 相似度similarity sig1.jaccard(sig2)print(f文档相似度估算: {similarity})## 性能优化通过NumPy实现NumPy版本仅使能NumPy SIMD能力相比于纯数组版本有一定性能提升import hashlibimport structimport timefrom typing import Listimport numpy as npdef compute_hash(s: str) - int:“”计算字符串的 64 位哈希值。“”hash_bytes hashlib.sha256(s.encode(‘utf-8’)).digest()return struct.unpack(‘Q’, hash_bytes[:8])[0]def compute_minhash_signature(shingles: List[str], a: List[int], b: List[int]) - List[int]:a_arr np.array(a, dtypenp.uint64)b_arr np.array(b, dtypenp.uint64)U64_MAX np.iinfo(np.uint64).maxsignature np.full(len(a), U64_MAX, dtypenp.uint64)for shingle in shingles:hash_value np.uint64(compute_hash(shingle))raw a_arr * hash_value b_arrcandidate np.where(raw U64_MAX, np.uint64(0), raw)signature np.minimum(signature, candidate)return signature.tolist()def main():# 性能测试配置 NUM_HASHES 1024 # 哈希函数数量NUM_SHINGLES 1000 # 文档 shingle 数量WARMUP_ITER 10 # 预热迭代次数BENCH_ITER 100 # 正式测试迭代次数# 生成测试数据a [1000 i * 137 for i in range(NUM_HASHES)]b [200 i * 251 for i in range(NUM_HASHES)]shingles [fshingle_{i:08d} for i in range(NUM_SHINGLES)]print(“” * 40)print(“Python NumPy MinHash Performance Benchmark”)print(“” * 40)print(“Configuration:”)print(f - Hash functions (len_a): {NUM_HASHES}“)print(f” - Shingles per doc: {NUM_SHINGLES}“)print(f” - Warmup iterations: {WARMUP_ITER}“)print(f” - Benchmark iterations: {BENCH_ITER}“)print()# 预热for _ in range(WARMUP_ITER):_ compute_minhash_signature(shingles, a, b)# 正式测试start time.perf_counter()for _ in range(BENCH_ITER):_ compute_minhash_signature(shingles, a, b)elapsed time.perf_counter() - starttotal_ops NUM_HASHES * NUM_SHINGLES * BENCH_ITERops_per_sec total_ops / elapsedprint(“Results:”)print(f” - Total time: {elapsed * 1000:.3f}ms)print(f - Per iteration: {elapsed * 1000 / BENCH_ITER:.3f}ms)print(f - Total operations: {total_ops} (hash * shingle * iter)“)print(f” - Throughput: {ops_per_sec:.2e} ops/sec)print()# 验证正确性小数据测试test_shingles [“the cat”, “cat sat”, “sat on”, “on the”, “the mat”]test_a [1187, 2143, 3541, 4327, 5003, 6211, 7109, 8293, 9001, 9973]test_b [211, 433, 701, 983, 1291, 1601, 1913, 2221, 2531, 2843]sig compute_minhash_signature(test_shingles, test_a, test_b)print(“Correctness check (small data):”)print(f Signature[0…5]: {sig[:5]})ifname “main”:main()## 性能优化直接通过SVE指令实现通过C的SVE实现SIMD在Python代码中通过load_library加载C优化版本的动态库Python代码实现如下import ctypesimport hashlibimport structimport osfrom pathlib import Pathfrom typing import Listimport numpy as npdef _load_library():“”“加载 C 共享库”“”lib_dir Path(file).parentlib_name “libminhash_sve.so”lib_path lib_dir / lib_nameif not lib_path.exists():raise FileNotFoundError(f共享库 {lib_path} 不存在。请先编译 C 代码:\nf g -O3 -marcharmv8-asve -shared -fPIC f-o {lib_name} minhash_sve.cpp)lib ctypes.CDLL(str(lib_path))# 定义函数签名# minhash_inner_scalarlib.minhash_inner_scalar.argtypes [ctypes.POINTER(ctypes.c_uint64), # actypes.POINTER(ctypes.c_uint64), # bctypes.POINTER(ctypes.c_uint64), # sigctypes.c_size_t, # nctypes.c_uint64, # hash_value]lib.minhash_inner_scalar.restype None# minhash_inner_svelib.minhash_inner_sve.argtypes [ctypes.POINTER(ctypes.c_uint64),ctypes.POINTER(ctypes.c_uint64),ctypes.POINTER(ctypes.c_uint64),ctypes.c_size_t,ctypes.c_uint64,]lib.minhash_inner_sve.restype Nonereturn lib# 全局加载库_lib Nonetry:_lib _load_library()except FileNotFoundError as e:print(f警告: {e})def compute_hash(s: str) - int:“”“计算字符串的 64 位哈希值”“”hash_bytes hashlib.sha256(s.encode(‘utf-8’)).digest()return struct.unpack(‘Q’, hash_bytes[:8])[0]def compute_minhash_signature_sve(shingles: List[str],a: List[int],b: List[int],use_sve: bool True) - List[int]:“”使用 C SVE/标量实现的 MinHash 签名计算“”if _lib is None:raise RuntimeError(“C 共享库未加载”)a_arr np.array(a, dtypenp.uint64)b_arr np.array(b, dtypenp.uint64)signature np.full(len(a), np.iinfo(np.uint64).max, dtypenp.uint64)# 获取 C 指针a_ptr a_arr.ctypes.data_as(ctypes.POINTER(ctypes.c_uint64))b_ptr b_arr.ctypes.data_as(ctypes.POINTER(ctypes.c_uint64))func _lib.minhash_inner_sve if use_sve else _lib.minhash_inner_scalarfor shingle in shingles:hash_value compute_hash(shingle)sig_ptr signature.ctypes.data_as(ctypes.POINTER(ctypes.c_uint64))func(a_ptr, b_ptr, sig_ptr, len(a), hash_value)return signature.tolist()def compute_minhash_signature(shingles: List[str],a: List[int],b: List[int]) - List[int]:“”默认使用 SVE 实现的 MinHash 签名计算“”return compute_minhash_signature_sve(shingles, a, b, use_sveTrue)# 测试代码ifname “main”:import time# 配置NUM_HASHES 1024NUM_SHINGLES 1000WARMUP_ITER 10BENCH_ITER 100a [1000 i * 137 for i in range(NUM_HASHES)]b [200 i * 251 for i in range(NUM_HASHES)]shingles [fshingle_{i:08d} for i in range(NUM_SHINGLES)]print(“” * 50)print(“Python C SVE MinHash Performance Benchmark”)print(“” * 50)print(fConfiguration:“)print(f” - Hash functions: {NUM_HASHES}“)print(f” - Shingles: {NUM_SHINGLES}“)print(f” - Warmup: {WARMUP_ITER}, Benchmark: {BENCH_ITER}“)print()# 预热for _ in range(WARMUP_ITER):_ compute_minhash_signature_sve(shingles, a, b, use_sveTrue)# 测试 SVE 版本start time.perf_counter()for _ in range(BENCH_ITER):sig_sve compute_minhash_signature_sve(shingles, a, b, use_sveTrue)elapsed_sve time.perf_counter() - start# 测试标量版本start time.perf_counter()for _ in range(BENCH_ITER):sig_scalar compute_minhash_signature_sve(shingles, a, b, use_sveFalse)elapsed_scalar time.perf_counter() - starttotal_ops NUM_HASHES * NUM_SHINGLES * BENCH_ITERprint(“Results:”)print(f” C SVE: {elapsed_sve * 1000:.3f}ms f({total_ops / elapsed_sve:.2e} ops/sec)“)print(f” C Scalar: {elapsed_scalar * 1000:.3f}ms f({total_ops / elapsed_scalar:.2e} ops/sec)“)print()# 验证正确性if sig_sve sig_scalar:print(“SVE 和标量版本结果一致”)else:print(“结果不一致”)diff_count sum(1 for x, y in zip(sig_sve, sig_scalar) if x ! y)print(f” 差异位置数: {diff_count}“)# 小数据验证test_shingles [“the cat”, “cat sat”, “sat on”, “on the”, “the mat”]test_a [1187, 2143, 3541, 4327, 5003, 6211, 7109, 8293, 9001, 9973]test_b [211, 433, 701, 983, 1291, 1601, 1913, 2221, 2531, 2843]sig compute_minhash_signature(test_shingles, test_a, test_b)print(f”\nCorrectness check: {sig[:5]})minhash_sve.cpp代码实现如下#include#includeextern “C” {// 标量回退实现void minhash_inner_scalar(const uint64_t* __restrict a,const uint64_t* __restrict b,uint64_t* __restrict sig,size_t n,uint64_t hash_value) {const uint64_t U64_MAX UINT64_MAX;for (size_t j 0; j n; j) {uint64_t raw a[j] * hash_value b[j];uint64_t candidate (raw U64_MAX) ? 0 : raw;if (candidate sig[j]) {sig[j] candidate;}}}// SVE 向量化实现#if defined(__ARM_FEATURE_SVE)#include arm_sve.hvoid minhash_inner_sve(const uint64_t* __restrict a,const uint64_t* __restrict b,uint64_t* __restrict sig,size_t n,uint64_t hash_value) {const svuint64_t v_hash svdup_u64(hash_value);const svuint64_t v_max svdup_u64(UINT64_MAX);const svuint64_t v_zero svdup_u64(0);size_t j 0;svbool_t pg;while (svptest_any(svptrue_b64(), (pg svwhilelt_b64(j, n)))) {// 加载 a[j…], b[j…], sig[j…]svuint64_t va svld1_u64(pg, a j);svuint64_t vb svld1_u64(pg, b j);svuint64_t vsig svld1_u64(pg, sig j);// raw a * hash bsvuint64_t vraw svmul_u64_x(pg, va, v_hash);vraw svadd_u64_x(pg, vraw, vb);// candidate (raw U64_MAX) ? 0 : rawsvbool_t eq_max svcmpeq_u64(pg, vraw, v_max);svuint64_t vcandidate svsel_u64(eq_max, v_zero, vraw);// sig min(sig, candidate)vsig svmin_u64_x(pg, vsig, vcandidate);// 写回svst1_u64(pg, sig j, vsig);j svcntd();}}#else// 无 SVE 支持时SVE 函数退化为标量实现void minhash_inner_sve(const uint64_t* __restrict a,const uint64_t* __restrict b,uint64_t* __restrict sig,size_t n,uint64_t hash_value) {minhash_inner_scalar(a, b, sig, n, hash_value);}#endif} // extern “C”
)
