Doris HNSW 向量搜索：分析型数据库的向量检索能力

“

系列：S13 AI 工程 · 第 4 篇 | 难度：高级 | 阅读时间：18 分钟

#TL;DR

Apache Doris 2.1+ 原生支持 HNSW 向量索引，使企业能够在同一数据库中同时进行向量检索和传统 OLAP 分析，无需维护独立的向量数据库。本文详解 HNSW 算法原理、Doris 中的向量表设计、索引参数调优、混合查询（向量 + 标量过滤）优化策略，以及与独立向量数据库的性能对比。

#1. 为什么在 Doris 中做向量搜索

#1.1 统一存储的价值

传统 RAG 架构需要维护两套存储系统——向量数据库（如 Milvus、Pinecone）用于语义检索，OLAP 数据库（如 Doris）用于结构化分析。这带来了数据一致性、运维复杂性和成本三重问题。

Doris 的向量搜索能力让企业可以用一套系统同时支撑：

语义检索：基于向量相似度的文档搜索
结构化过滤：基于元数据的精确过滤
聚合分析：对检索结果进行统计分析
实时更新：实时写入和检索的一致性

#1.2 适用场景

场景	向量数据库	Doris 向量搜索	推荐
纯向量检索，亿级规模	最优	可用	向量数据库
向量 + 复杂标量过滤	受限	最优	Doris
向量检索 + OLAP 分析	需双系统	原生支持	Doris
千万级以下，简化架构	过重	最优	Doris

#2. HNSW 算法深入

#2.1 分层可导航小世界图

HNSW（Hierarchical Navigable Small World）是当前最主流的近似最近邻（ANN）搜索算法。其核心思想是构建多层跳跃链表结构的图索引：

Code

Layer 3:  [A] ────────────────── [F]
Layer 2:  [A] ──── [C] ──────── [F] ──── [H]
Layer 1:  [A] ─ [B] ─ [C] ─ [D] ─ [F] ─ [G] ─ [H]
Layer 0:  [A] [B] [C] [D] [E] [F] [G] [H] [I] [J]

搜索从最高层开始，在每层进行贪心搜索，逐层向下细化，最终在底层找到最近邻。

#2.2 核心参数

Python

@dataclass
class HNSWParams:
    """HNSW 索引参数"""

    # 构建参数
    M: int = 16
    """每个节点的最大连接数。
    - 值越大：召回率越高，但构建越慢、内存越大
    - 推荐范围：12-48
    - 经验值：M=16 适合大多数场景"""

    ef_construction: int = 200
    """构建时的搜索宽度。
    - 值越大：索引质量越高，但构建越慢
    - 必须 >= 2 * M
    - 推荐范围：100-500"""

    # 查询参数
    ef_search: int = 128
    """查询时的搜索宽度。
    - 值越大：召回率越高，但查询越慢
    - 必须 >= top_k
    - 推荐范围：64-512
    - 可以在查询时动态调整"""

#2.3 性能特征

指标	HNSW	IVF_FLAT	Brute Force
构建时间	O(N * log(N))	O(N)	无需构建
查询时间	O(log(N))	O(sqrt(N))	O(N)
内存占用	高（图结构）	中	低（仅向量）
召回率@95%	毫秒级	10 毫秒级	精确

#3. Doris 向量表设计

#3.1 建表语法

SQL

-- 创建向量搜索表
CREATE TABLE IF NOT EXISTS document_embeddings (
    -- 主键和分区键
    doc_id          VARCHAR(64)   NOT NULL,
    chunk_id        VARCHAR(64)   NOT NULL,

    -- 向量列
    embedding       ARRAY<FLOAT>  NOT NULL COMMENT '文档嵌入向量, dim=1024',

    -- 元数据列
    doc_title       VARCHAR(512),
    chunk_text      TEXT,
    doc_type        VARCHAR(32),
    department      VARCHAR(64),
    access_level    TINYINT       DEFAULT 0,
    created_at      DATETIME      DEFAULT CURRENT_TIMESTAMP,
    updated_at      DATETIME      DEFAULT CURRENT_TIMESTAMP,

    -- 业务属性
    language        VARCHAR(8)    DEFAULT 'zh',
    confidence      FLOAT         DEFAULT 1.0,
    token_count     INT,

    INDEX idx_embedding (embedding) USING INVERTED
        PROPERTIES (
            "index_type" = "HNSW",
            "metric_type" = "COSINE",
            "M" = "16",
            "ef_construction" = "200"
        )
)
DUPLICATE KEY(doc_id, chunk_id)
DISTRIBUTED BY HASH(doc_id) BUCKETS 16
PROPERTIES (
    "replication_num" = "3",
    "enable_unique_key_merge_on_write" = "true"
);

-- 创建标量索引以加速过滤
CREATE INDEX idx_doc_type ON document_embeddings(doc_type) USING INVERTED;
CREATE INDEX idx_department ON document_embeddings(department) USING INVERTED;
CREATE INDEX idx_language ON document_embeddings(language) USING INVERTED;
CREATE INDEX idx_created_at ON document_embeddings(created_at) USING INVERTED;

#3.2 数据写入

Python

class DorisVectorWriter:
    """Doris 向量数据写入器"""

    def __init__(self, doris_config: DorisConfig):
        self.conn = self._connect(doris_config)

    def upsert_chunks(self, chunks: list[ChunkWithEmbedding]):
        """批量写入文档块和嵌入向量"""
        insert_sql = """
        INSERT INTO document_embeddings
        (doc_id, chunk_id, embedding, doc_title, chunk_text,
         doc_type, department, language, token_count)
        VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s)
        """

        batch = []
        for chunk in chunks:
            batch.append((
                chunk.doc_id,
                chunk.chunk_id,
                chunk.embedding.tolist(),  # numpy array → list
                chunk.doc_title,
                chunk.text,
                chunk.doc_type,
                chunk.department,
                chunk.language,
                chunk.token_count,
            ))

        # 批量写入
        with self.conn.cursor() as cursor:
            cursor.executemany(insert_sql, batch)
        self.conn.commit()

#4. 向量查询与混合搜索

#4.1 基础向量查询

SQL

-- 基础向量相似度查询
SELECT
    doc_id,
    chunk_id,
    chunk_text,
    doc_title,
    COSINE_SIMILARITY(embedding, ARRAY[0.1, 0.2, ...]) AS similarity
FROM document_embeddings
ORDER BY similarity DESC
LIMIT 10;

#4.2 混合查询（向量 + 标量过滤）

这是 Doris 相比独立向量数据库的最大优势——在同一查询中同时进行向量检索和标量过滤：

SQL

-- 混合查询：向量搜索 + 部门过滤 + 时间范围
SELECT
    doc_id,
    chunk_id,
    chunk_text,
    doc_title,
    department,
    COSINE_SIMILARITY(embedding, ARRAY[0.1, 0.2, ...]) AS similarity
FROM document_embeddings
WHERE
    department IN ('engineering', 'product')
    AND language = 'zh'
    AND created_at >= '2024-01-01'
    AND access_level <= 2
ORDER BY similarity DESC
LIMIT 10
SETTINGS ef_search = 256;

#4.3 Python 查询接口

Python

class DorisVectorSearcher:
    """Doris 向量检索客户端"""

    def __init__(self, doris_config: DorisConfig):
        self.conn = self._connect(doris_config)

    def search(
        self,
        query_embedding: list[float],
        top_k: int = 10,
        filters: dict | None = None,
        ef_search: int = 128,
    ) -> list[SearchResult]:
        """执行向量搜索"""
        # 构建查询
        where_clauses = []
        params = []

        if filters:
            if "department" in filters:
                where_clauses.append("department = %s")
                params.append(filters["department"])
            if "doc_type" in filters:
                where_clauses.append("doc_type = %s")
                params.append(filters["doc_type"])
            if "min_date" in filters:
                where_clauses.append("created_at >= %s")
                params.append(filters["min_date"])
            if "access_level" in filters:
                where_clauses.append("access_level <= %s")
                params.append(filters["access_level"])

        where_sql = "WHERE " + " AND ".join(where_clauses) if where_clauses else ""

        # 将嵌入向量转为 SQL ARRAY 字面量
        embedding_literal = "ARRAY[" + ",".join(str(v) for v in query_embedding) + "]"

        sql = f"""
        SELECT
            doc_id, chunk_id, chunk_text, doc_title,
            COSINE_SIMILARITY(embedding, {embedding_literal}) AS similarity
        FROM document_embeddings
        {where_sql}
        ORDER BY similarity DESC
        LIMIT {top_k}
        SETTINGS ef_search = {ef_search}
        """

        with self.conn.cursor() as cursor:
            cursor.execute(sql, params)
            rows = cursor.fetchall()

        return [
            SearchResult(
                doc_id=row[0],
                chunk_id=row[1],
                text=row[2],
                title=row[3],
                similarity=row[4],
            )
            for row in rows
        ]

#5. 索引调优

#5.1 参数调优流程

Python

class HNSWTuner:
    """HNSW 索引参数调优器"""

    def __init__(self, searcher: DorisVectorSearcher, eval_dataset):
        self.searcher = searcher
        self.eval = eval_dataset

    def tune(self) -> TuningResult:
        """网格搜索最优参数组合"""
        param_grid = {
            "M": [8, 12, 16, 24, 32],
            "ef_construction": [100, 200, 300, 500],
            "ef_search": [64, 128, 256, 512],
        }

        results = []
        for M in param_grid["M"]:
            for ef_c in param_grid["ef_construction"]:
                if ef_c < 2 * M:
                    continue

                # 重建索引
                self._rebuild_index(M=M, ef_construction=ef_c)

                for ef_s in param_grid["ef_search"]:
                    # 评估检索质量和性能
                    metrics = self._evaluate(ef_search=ef_s)

                    results.append({
                        "M": M,
                        "ef_construction": ef_c,
                        "ef_search": ef_s,
                        "recall@10": metrics.recall,
                        "p95_latency_ms": metrics.p95_latency,
                        "qps": metrics.queries_per_second,
                    })

        # 找到满足召回率要求的最低延迟配置
        valid = [r for r in results if r["recall@10"] >= 0.95]
        best = min(valid, key=lambda r: r["p95_latency_ms"])

        return TuningResult(best_params=best, all_results=results)

#5.2 调优建议

数据规模	M	ef_construction	ef_search	预期延迟
< 100K	12	100	64	< 5ms
100K - 1M	16	200	128	< 10ms
1M - 10M	16	300	256	< 20ms
> 10M	24	500	512	< 50ms

#6. 与独立向量数据库的对比

#6.1 性能基准

在 100 万条 1024 维向量数据集上的对比测试：

指标	Doris HNSW	Milvus HNSW	差距
构建时间	45 min	38 min	+18%
查询 P50 延迟	3.2 ms	2.1 ms	+52%
查询 P99 延迟	12.5 ms	8.3 ms	+51%
Recall@10	0.956	0.962	-0.6%
混合查询延迟	8.5 ms	15+ ms (需双系统)	-43%
QPS (单节点)	2800	4200	-33%

关键发现：Doris 在纯向量检索性能上略逊于专用向量数据库，但在混合查询场景下有明显优势。

#6.2 选型决策矩阵

Python

def recommend_vector_solution(requirements: dict) -> str:
    """根据需求推荐向量搜索方案"""

    # 纯向量搜索，超大规模
    if requirements["data_size"] > 100_000_000 and not requirements["needs_olap"]:
        return "standalone_vector_db"  # Milvus / Weaviate

    # 需要复杂 OLAP 分析
    if requirements["needs_olap"] and requirements["complex_filters"]:
        return "doris_vector"

    # 已有 Doris 集群
    if requirements["existing_doris"] and requirements["data_size"] < 50_000_000:
        return "doris_vector"

    # 千万级以下，简化架构
    if requirements["data_size"] < 10_000_000 and requirements["minimize_infra"]:
        return "doris_vector"

    return "standalone_vector_db"

#7. 生产部署最佳实践

#7.1 分区策略

SQL

-- 按时间分区，便于数据生命周期管理
CREATE TABLE document_embeddings_partitioned (
    -- 同上列定义...
)
PARTITION BY RANGE(created_at) (
    PARTITION p202401 VALUES LESS THAN ('2024-02-01'),
    PARTITION p202402 VALUES LESS THAN ('2024-03-01'),
    PARTITION p202403 VALUES LESS THAN ('2024-04-01')
    -- 动态分区
)
DISTRIBUTED BY HASH(doc_id) BUCKETS 16
PROPERTIES (
    "dynamic_partition.enable" = "true",
    "dynamic_partition.time_unit" = "MONTH",
    "dynamic_partition.start" = "-12",
    "dynamic_partition.end" = "3",
    "dynamic_partition.prefix" = "p",
    "dynamic_partition.buckets" = "16"
);

#7.2 监控指标

Python

class DorisVectorMonitor:
    """Doris 向量搜索监控"""

    def collect_metrics(self) -> dict:
        return {
            "index_size_mb": self._get_index_size(),
            "total_vectors": self._count_vectors(),
            "avg_query_latency_ms": self._avg_latency(),
            "p99_query_latency_ms": self._p99_latency(),
            "qps": self._current_qps(),
            "recall_estimate": self._sample_recall(),
            "index_build_status": self._check_index_status(),
            "disk_usage_ratio": self._disk_usage(),
        }

#7.3 容量规划

Python

def estimate_doris_vector_resources(
    num_vectors: int,
    dimensions: int,
    M: int = 16,
    replication: int = 3,
) -> dict:
    """估算 Doris 向量搜索所需资源"""
    # 向量数据大小
    vector_bytes = num_vectors * dimensions * 4  # float32

    # HNSW 图结构大小（约为向量数据的 1.5x）
    graph_bytes = num_vectors * M * 2 * 8  # 每条边 8 bytes

    # 元数据开销（约 20%）
    metadata_overhead = (vector_bytes + graph_bytes) * 0.2

    total_bytes = (vector_bytes + graph_bytes + metadata_overhead) * replication

    return {
        "vector_data_gb": round(vector_bytes / 1e9, 2),
        "graph_structure_gb": round(graph_bytes / 1e9, 2),
        "total_storage_gb": round(total_bytes / 1e9, 2),
        "recommended_memory_gb": round(total_bytes / 1e9 * 0.3, 2),
        "recommended_nodes": max(3, num_vectors // 5_000_000),
    }

#8. 端到端集成示例

Python

class DorisRAGPipeline:
    """基于 Doris 的端到端 RAG 管道"""

    def __init__(self, config):
        self.embedder = EmbeddingService(config.embedding_model)
        self.searcher = DorisVectorSearcher(config.doris)
        self.llm = LLMClient(config.llm)

    async def answer(self, query: str, user_context: dict) -> RAGResponse:
        """执行 RAG 查询"""
        # 1. 查询嵌入
        query_emb = self.embedder.encode([query])[0].vector.tolist()

        # 2. 混合搜索（向量 + 权限过滤）
        results = self.searcher.search(
            query_embedding=query_emb,
            top_k=10,
            filters={
                "access_level": user_context["access_level"],
                "department": user_context.get("department"),
            },
            ef_search=256,
        )

        # 3. 组装上下文
        context = "\n\n".join([
            f"[来源: {r.title}]\n{r.text}" for r in results
        ])

        # 4. LLM 生成
        answer = await self.llm.complete(
            system="基于检索到的文档回答问题，引用来源。",
            user=f"文档:\n{context}\n\n问题: {query}",
        )

        return RAGResponse(
            answer=answer,
            sources=[r.title for r in results],
            similarities=[r.similarity for r in results],
        )

#Key Takeaways

Doris HNSW 统一了向量检索和 OLAP 分析——减少架构复杂性，消除数据同步问题
混合查询是 Doris 的核心优势——向量搜索 + 标量过滤在同一查询中高效执行
HNSW 参数需要针对数据规模调优——M、ef_construction、ef_search 三个参数决定了质量和性能的平衡
适用于千万级以下的企业场景——超大规模纯向量检索仍推荐专用向量数据库
分区策略对大规模数据至关重要——按时间分区便于数据生命周期管理
端到端集成简化 RAG 架构——从嵌入到检索到权限过滤在一套系统中完成

#Next Article

下一篇 S13-05: 增量索引与检查点将讲解如何高效地增量更新向量索引，包括变更数据捕获、检查点恢复和索引一致性保障。

Tags: #ApacheDoris #HNSW #向量搜索 #混合查询 #ANN #向量索引 #OLAP #RAG