如何将图像哈希相似度SQL查询转为Python Elasticsearch DSL实现？

问题背景 #

原有SQL通过bit_count和异或运算计算图像哈希的汉明距离与相似度索引，现需迁移到1000万级数据的Elasticsearch环境，使用Python的Elasticsearch DSL实现时遇到三个问题：

• 无法获取imagename字段
• 无法筛选相似度≥75%的结果
• 长哈希值读取时返回负数

逐个问题解决 #

1. 无法获取imagename字段 #

script_fields仅返回脚本计算的字段，不会自动返回原始文档字段。需在Search请求中显式指定要返回的原始字段：

• 使用.source()方法指定字段，例如.source(['imagename'])
• 或开启fetch_source=True并指定字段列表，确保imagename被包含在返回结果中

2. 筛选simindex≥75%的结果 #

不要在Python端过滤script_fields结果，直接在Elasticsearch查询阶段用script_score结合min_score实现（与Kibana逻辑一致）：

• 用script_score将相似度索引设为文档的_score
• 设置min_score=0.75（对应SQL中100*(...)≥75的比例值）
• 可额外添加哈希长度匹配条件，建议提前在映射中新增hash_length字段并索引，用term查询过滤，避免全量脚本计算

3. 长哈希值返回负数 #

Elasticsearch默认long类型为有符号64位，哈希值超过2^63-1时会溢出为负数，两种解决方式：

• 推荐方案：修改Elasticsearch映射，将hash字段类型设为unsigned_long（ES 7.0+支持），直接读取无符号数值
• 兼容方案：在Painless脚本中通过位运算转换为无符号值：def dbHash = doc['hash'].value & 0xFFFFFFFFFFFFFFFFL;

完整Python代码实现 #

from elasticsearch import Elasticsearch
from elasticsearch_dsl import Search

# 初始化ES客户端
client = Elasticsearch("your_es_host:port")

# 创建汉明距离计算脚本（仅需执行一次）
def create_hamming_script():
    script_body = {
        "script": {
            "lang": "painless",
            "source": """
                def targetHash = params.targetHash;
                def dbHash = doc['hash'].value & 0xFFFFFFFFFFFFFFFFL; // 处理无符号转换
                def hammingDistance = 0;
                for (int i = 0; i < 64; i++) {
                    if (((dbHash >> i) & 1) != ((targetHash >> i) & 1)) {
                        hammingDistance++;
                    }
                }
                if (dbHash == targetHash) {
                    return 1.0;
                }
                def similarityIndex = 1 - (hammingDistance / 64.0);
                return similarityIndex;
            """
        }
    }
    client.put_script(id="hammingdistance", body=script_body)

# 构建相似图像查询
def search_similar_images(target_hash_decimal):
    s = Search(using=client, index="hash")
    
    # 脚本评分查询，将相似度作为文档score
    s = s.query(
        "script_score",
        query={"match_all": {}},
        script={
            "id": "hammingdistance",
            "params": {"targetHash": target_hash_decimal}
        }
    )
    
    # 筛选相似度≥75%的结果
    s = s.extra(min_score=0.75)
    
    # 指定返回字段
    s = s.source(["imagename", "hash"])
    
    # 按相似度降序排序
    s = s.sort({"_score": {"order": "desc"}})
    
    # 执行查询
    return s.execute()

# 示例调用
if __name__ == "__main__":
    create_hamming_script()
    # 目标哈希（0x1800387EE3414302的十进制值）
    target_hash = 0x1800387EE3414302
    results = search_similar_images(target_hash)
    
    # 处理返回结果
    for hit in results:
        imagename = hit.imagename
        hash_hex = hex(hit.hash)
        similarity_percent = hit.meta.score * 100
        print(f"图像名称: {imagename}, 哈希值: {hash_hex}, 相似度: {similarity_percent:.2f}%")

性能优化建议 #

• 提前计算并索引hash_length字段，在查询中添加term过滤：

  s = s.query("bool", filter=[
      {"term": {"hash_length": len(hex(target_hash)) - 2}}  # 去除hex()返回的0x前缀
  ])
  

• 针对千万级数据，可结合Elasticsearch的HNSW近似最近邻搜索优化哈希查询性能

内容的提问来源于stack exchange，提问作者Dhruv Dugar