在Databricks中实现向量搜索

搜索一直是分析的核心。无论是追踪正确的交易、筛选客户记录还是提取特定评论，传统方法通常都是关键词搜索。当您确切知道要查找什么时，关键词搜索简单有效，但当语言混乱、模糊或含义比确切词语更重要时，这种方法很快就会失效。这就是向量搜索改变游戏规则的地方。

向量搜索不匹配字面关键词，而是依赖于嵌入——文本、图像或其他非结构化内容的高维数值表示——这些表示捕获了语义含义。在向量空间中，两个含义相同的文本片段即使使用完全不同的词语，它们的嵌入也会彼此接近。

实际示例

例如：

客户评论说"配送花了很长时间"和另一条说"运输很慢"在语义上是相似的。基于关键词的搜索可能会错过这种联系，但向量搜索会同时显示两者。

直到最近，实现向量搜索还意味着要将多个组件拼接在一起：在Python中运行嵌入模型，将向量存储在外部向量数据库中，同步元数据，以及编写自定义API进行检索。这很复杂、脆弱，并且几乎总是存在于受治理的数据仓库之外，使得治理、沿袭和合规性变得更加困难。

Databricks将原生向量搜索引入Lakehouse。通过Python笔记本和SQL命令，您可以生成嵌入、构建向量索引并执行语义查询。

实施背景

Bakehouse是Databricks中为虚构面包店提供的示例数据集。它接收来自许多特许经营店和日期的快速增长的自由格式客户评论。由于反馈使用各种措辞（同义词、俚语、拼写错误、多语言提示），依赖简单关键词过滤的分析师很难一致地找到语义相似的投诉（例如，“未收到退款”、“钱没回来”、“退货待处理”）。因此，像延迟交付、新鲜度问题或退款摩擦这样的根本原因模式被检测得很晚，处理是手动且耗时的，特许经营店级别的责任划分不均匀。

我们需要一个受治理、低延迟的语义搜索功能来覆盖评论，以便利益相关者可以用自然语言提问，并按特许经营店和时间切片结果——无需为每个用例构建自定义NLP管道。

核心组件

Delta表 bakehouse.reviews.customer_feedback，包含列：new_id（主键）、review（字符串）、franchiseID、review_date（日期）
向量搜索端点：托管计算，承载一个或多个索引。选择标准（低延迟）或存储优化（十亿级规模，每向量成本更低，延迟更高）
Delta同步索引（托管嵌入）：使用托管模型（例如databricks-bge-large-en）从review计算嵌入，并通过CDF与表更改保持同步
查询：Databricks SQL vector_search(index => ..., query_text => ..., num_results => N)

实施步骤

步骤1：启用变更数据馈送

标准向量搜索端点需要启用CDF，以便索引可以增量同步。

1
2


ALTER TABLE bakehouse.reviews.customer_feedback
SET TBLPROPERTIES (delta.enableChangeDataFeed = true);

步骤2：创建/确保端点和Delta同步索引（托管嵌入）

创建笔记本并运行以下代码：

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45


%pip install -q databricks-vectorsearch
dbutils.library.restartPython()

from databricks.vector_search.client import VectorSearchClient

CATALOG = "bakehouse"
SCHEMA  = "reviews"
TABLE   = "customer_feedback"
SRC_FQN = f"{CATALOG}.{SCHEMA}.{TABLE}"

ENDPOINT_NAME = "bakehouse-vs-endpoint"
INDEX_FQN     = f"{CATALOG}.{SCHEMA}.customer_feedback_idx_v2"
PRIMARY_KEY   = "new_id"
TEXT_COL      = "review"
EMBEDDING_MODEL_ENDPOINT = "databricks-bge-large-en"

# 1) 先决条件：在源Delta表上启用变更数据馈送
spark.sql(f"""
ALTER TABLE {SRC_FQN}
SET TBLPROPERTIES (delta.enableChangeDataFeed = true)
""")

# 2) 创建/确保向量搜索端点
vsc = VectorSearchClient()
try:
   vsc.get_endpoint(ENDPOINT_NAME)
   print(f"Endpoint exists: {ENDPOINT_NAME}")
except Exception:
   print(f"Creating endpoint: {ENDPOINT_NAME}")
   vsc.create_endpoint(name=ENDPOINT_NAME, endpoint_type="STANDARD")
   vsc.create_endpoint_and_wait(name=ENDPOINT_NAME)
   print("Endpoint ONLINE.")

# 3) 创建从文本计算嵌入的Delta同步索引
print(f"Creating index with computed embeddings: {INDEX_FQN}")
idx = vsc.create_delta_sync_index_and_wait(
   endpoint_name=ENDPOINT_NAME,
   index_name=INDEX_FQN,
   primary_key=PRIMARY_KEY,
   source_table_name=SRC_FQN,
   pipeline_type="TRIGGERED",
   embedding_source_column=TEXT_COL,
   embedding_model_endpoint_name=EMBEDDING_MODEL_ENDPOINT
)
print("Index ONLINE:", INDEX_FQN)

步骤3：创建语义查询

运行语义查询检查退款问题：

1
2
3
4
5
6
7


SELECT new_id, review, franchiseID, review_date, score
FROM vector_search(
   index => 'bakehouse.reviews.customer_feedback_idx_v2',
   query_text => 'refund not received',
   num_results => 10
)
ORDER BY score DESC;

检查交付延迟的语义查询：

1
2
3
4
5
6


SELECT new_id, review, franchiseID, review_date, search_score
FROM vector_search(
   index        => 'bakehouse.reviews.customer_feedback_idx_v2',
   query_text   => 'delivery delays and late arrivals',
   num_results  => 50
)

业务价值

通过此实现，我们让业务团队能够用自然语言提问，而无需ML管道。业务可以自助服务以下场景：

退款：区域经理输入"未收到退款"并按franchiseID过滤→无需等待新的ML标签器即可按日期查看峰值模式
新鲜度：质量负责人搜索"面包不新鲜"→将问题聚类到供应商；开启有针对性的纠正措施
交付：运营分析师查询"延迟到达/交付延迟"→在承诺预算前验证路由变更

总结

我们通过将Unity Catalog Delta表与Databricks向量搜索端点和从review列直接计算嵌入的Delta同步索引（CDF开启）配对，在Bakehouse客户评论上实现了语义搜索。这让分析师可以用简单的英语查询，并立即在各个特许经营店和日期之间浮现含义匹配的评论。托管索引消除了为每个用例构建和维护一次性ML分类器的需要，通过连续/触发同步保持结果新鲜，并在UC下保持治理。

要将此演进为RAG，请使用相同的索引作为检索器（通过vector_search按前K条评论），并将这些片段传递给在Databricks上服务的LLM，指示仅从检索到的文本中回答并引用new_id（例如，[12345]）。这在搜索层之上增加了简洁、有根据的摘要和建议，而无需更改数据或治理模型。