Python实用工具：高效数据 sketch 工具 datasketch 深度解析

pip install datasketch

git clone https://github.com/ekzhu/datasketch.git
cd datasketch
python setup.py install

from datasketch import MinHash

# 定义两个文本集合（单词列表）
text1 = "python is a powerful programming language".split()
text2 = "python is an easy-to-learn programming language".split()

# 初始化 MinHash 对象，设置哈希函数数量（此处为 128 个）
m1 = MinHash(num_perm=128)
m2 = MinHash(num_perm=128)

# 向 MinHash 对象中添加元素（需转换为字节串）
for word in text1:
    m1.update(word.encode('utf-8'))
for word in text2:
    m2.update(word.encode('utf-8'))

# 计算 Jaccard 相似度估计值
jaccard_sim = m1.jaccard(m2)
print(f"Jaccard 相似度估计：{jaccard_sim:.4f}")

# 生成 MinHash 签名（可用于存储或传输）
m1_signature = m1.digest()
m2_signature = m2.digest()

from datasketch import MinHash, MinHashLSHForest

# 生成多个文档的 MinHash 签名
docs = [
    "apple banana orange".split(),
    "apple banana grape".split(),
    "pear pineapple orange".split(),
    "grape melon pear".split()
]
minhashes = []
for doc in docs:
    m = MinHash(num_perm=128)
    for word in doc:
        m.update(word.encode('utf-8'))
    minhashes.append(m)

# 初始化 LSH 森林并添加签名
forest = MinHashLSHForest(num_perm=128)
for i, m in enumerate(minhashes):
    forest.add(i, m)
forest.index()  # 构建索引

# 查询与第一个文档相似的项（阈值设为 0.5）
query_m = minhashes[0]
result = forest.query(query_m, 0.5)
print("相似文档索引：", result)  # 输出可能包含 0（自身）、1 等

from datasketch import HyperLogLog

# 模拟用户日志（百万级用户 ID）
import random
user_ids = [random.randint(1, 10**6) for _ in range(10**5)]  # 10 万条日志，真实唯一用户约 8 万

# 初始化 HyperLogLog，设置分桶数（2^14 = 16384 桶，内存约 16KB）
hll = HyperLogLog(p=14)  # p 决定桶数，p=14 对应 2^14 桶

for user_id in user_ids:
    hll.update(str(user_id).encode('utf-8'))

# 估算基数与真实值对比
estimated_count = hll.count()
true_count = len(set(user_ids))
print(f"估计唯一用户数：{estimated_count}")
print(f"真实唯一用户数：{true_count}")

from datasketch import HyperLogLog

# 模拟三个节点的 HyperLogLog
hll1 = HyperLogLog(p=14)
hll2 = HyperLogLog(p=14)
hll3 = HyperLogLog(p=14)

# 各节点更新数据
for i in range(1, 30001):
    hll1.update(f"user_{i}".encode('utf-8'))
for i in range(20001, 50001):
    hll2.update(f"user_{i}".encode('utf-8'))
for i in range(40001, 70001):
    hll3.update(f"user_{i}".encode('utf-8'))

# 合并节点结果
merged_hll = HyperLogLog(p=14)
merged_hll.merge(hll1)
merged_hll.merge(hll2)
merged_hll.merge(hll3)

# 估算总基数（真实唯一用户为 70000 - 1 = 69999，因区间重叠）
print("合并后估计基数：", merged_hll.count())

from datasketch import CountMinSketch

# 初始化 Count-Min Sketch，设置哈希函数数（k=4）和草图行数（w=1024）
cms = CountMinSketch(k=4, w=1024)

# 模拟日志流：单词列表
log_stream = ["apple", "banana", "apple", "orange", "banana", "apple", "grape"]

for word in log_stream:
    cms.add(word, 1)  # 添加元素，计数加 1

# 查询单词频率
print("apple 估计频率：", cms.query("apple"))
print("banana 估计频率：", cms.query("banana"))
print("grape 估计频率：", cms.query("grape"))

from datasketch import CountMinSketch

# 初始化两个 Count-Min Sketch
cms1 = CountMinSketch(k=4, w=1024)
cms2 = CountMinSketch(k=4, w=1024)

# 数据流 1：用户点击商品 A、B、C
cms1.add("A", 1)
cms1.add("B", 1)
cms1.add("C", 1)

# 数据流 2：用户点击商品 B、C、D
cms2.add("B", 1)
cms2.add("C", 1)
cms2.add("D", 1)

# 估算交集大小（真实交集为 B、C，计数均为 1）
intersection_estimate = cms1.intersection(cms2)
print("交集大小估计：", intersection_estimate)  # 可能输出 2 或相近值

from datasketch import HyperLogLog
import time

# 模拟用户浏览日志生成（用户 ID、时间戳、商品 ID）
def generate_logs(num_logs):
    for _ in range(num_logs):
        user_id = f"user_{random.randint(1, 10**5)}"
        yield user_id.encode('utf-8'), time.time()

# 初始化 HyperLogLog（p=16，内存约 64KB，误差约 1%）
hll = HyperLogLog(p=16)

# 模拟实时日志处理
for user_id, timestamp in generate_logs(10000):
    hll.update(user_id)
    # 此处可添加时间窗口逻辑（如每小时合并一次）

# 每日结束时输出活跃用户估计值
daily_active_users = hll.count()
print(f"今日活跃用户估计：{daily_active_users}")

from datasketch import MinHash, MinHashLSHForest

# 假设已收集各商品的浏览用户列表（商品 ID: 用户集合）
product_users = {
    "P001": {"user_1", "user_2", "user_3", "user_4"},
    "P002": {"user_2", "user_3", "user_5"},
    "P003": {"user_4", "user_6", "user_7"},
    "P004": {"user_3", "user_4", "user_7", "user_8"}
}

# 生成商品 MinHash 签名
minhash_dict = {}
for pid, users in product_users.items():
    m = MinHash(num_perm=128)
    for user in users:
        m.update(user.encode('utf-8'))
    minhash_dict[pid] = m

# 构建 LSH 森林
forest = MinHashLSHForest(num_perm=128)
for pid, m in minhash_dict.items():
    forest.add(pid, m)
forest.index()

# 为商品 P001 推荐相似商品（阈值 0.5）
query_pid = "P001"
query_m = minhash_dict[query_pid]
similar_products = forest.query(query_m, 0.5)
print(f"与 {query_pid} 相似的商品：{similar_products}")  # 可能返回 P002、P004 等

from datasketch import CountMinSketch

# 商品类别映射（假设商品 ID 前两位为类别代码）
product_categories = {
    "P001": "CL01",
    "P002": "CL02",
    "P003": "CL01",
    "P004": "CL03",
    "P005": "CL02"
}

# 初始化 Count-Min Sketch，设置哈希函数数（k=6）和草图行数（w=2048）
cms = CountMinSketch(k=6, w=2048)

# 模拟用户浏览日志（包含商品 ID）
browse_logs = ["P001", "P002", "P003", "P004", "P002", "P001", "P005", "P002"]

for product_id in browse_logs:
    category = product_categories[product_id]
    cms.add(category, 1)  # 统计对应类别的浏览次数

# 查询高频类别
categories = list(set(product_categories.values()))
for category in categories:
    estimated_count = cms.query(category)
    print(f"{category} 浏览次数估计: {estimated_count}")

# 找出浏览次数最高的类别
top_category = max(categories, key=lambda x: cms.query(x))
print(f"浏览次数最高的类别: {top_category}")