当前位置：首页 > news >正文

Milvus：Json字段详解（十）

news 2025/11/5 8:13:36

一、JSON 字段概述

1.1 什么是 JSON 字段

Milvus 的 JSON 字段是支持结构化键值数据存储的灵活数据类型，能兼容嵌套对象、数组等复杂结构，同时提供索引和查询优化能力，适用于多类复杂数据场景。以下是完善后的详细指南，包含更细致的说明和代码解析。

JSON 字段是 Milvus 中以 DataType.JSON 声明的 Schema 字段类型，专门用于存储结构化及半结构化数据。它打破了传统数据库刚性列的限制，允许在单个字段中包含嵌套对象（如多层级的供应商信息）、数组（如产品标签列表）和混合数据类型（字符串、数值、布尔值共存），同时支持针对性索引配置，确保复杂结构数据的查询效率。

1.2 应用场景

产品目录系统：存储不同品类产品的差异化属性，比如电子产品的参数、服装的尺码信息，无需为每类产品单独设计 Schema。
内容管理系统：管理带复杂嵌套结构的文档属性，例如文章的作者信息、分类标签、关联附件等多层级数据。
用户偏好引擎：记录随时间变化的用户行为数据，比如用户浏览历史、收藏偏好、设置选项等动态信息。
物联网数据存储：处理设备传感器产生的异构数据，比如温度、湿度等数值型数据与设备位置、状态等结构化数据的混合存储。

二、JSON 字段基础

2.1 JSON 字段结构示例

"metadata": {"category": "electronics",  // 一级字段：产品类别（字符串类型）"brand": "BrandA",          // 一级字段：品牌（字符串类型）"in_stock": true,           // 一级字段：库存状态（布尔类型）"price": 99.99,             // 一级字段：价格（数值类型）"string_price": "99.99",    // 一级字段：字符串格式价格（字符串类型）"tags": ["clearance", "summer_sale"],  // 一级字段：标签数组（字符串数组）"supplier": {               // 一级字段：嵌套对象（供应商信息）"name": "SupplierX",      // 二级嵌套字段：供应商名称"country": "USA",         // 二级嵌套字段：供应商国家"contact": {              // 三级嵌套字段：联系信息子对象"email": "support@supplierx.com",  // 三级字段：邮箱"phone": "+1-800-555-0199"         // 三级字段：电话}}
}

结构特点：支持多层嵌套（示例中最多三级）、数组与对象混用、多数据类型共存，且字段可灵活扩展。

2.2 JSON 字段 vs 动态字段

特征	JSON 字段	动态字段
Schema 定义	标量字段，需在 Collections Schema 中明确声明为 DataType.JSON	隐藏的 JSON 字段（默认名为 #meta ），无需手动声明
使用情况	适用于存储模式已知、结构一致的结构化数据，比如固定属性的产品信息	适用于存储模式灵活、频繁变化或半结构化数据，比如用户自定义属性
控制方式	字段名称、结构由开发者完全控制，可提前规划数据组织形式	未定义的字段由系统自动管理，无需手动维护字段结构
查询方式	需通过字段名 + JSON 路径查询（如 metadata["category"] ）	两种查询方式：直接用动态键（如 "dynamic_key" ）或通过 #meta 路径（如 #meta["dynamic_key"] ）
核心优势	结构可控、查询路径明确，支持针对性索引优化	无需预先定义结构，适配快速变化的数据场景
适用场景	产品目录、固定格式文档存储	用户自定义属性、临时数据记录

三、基本操作

3.1 定义 JSON 字段

# 1. 导入必要的库：MilvusClient 用于连接服务器，DataType 定义字段类型
from pymilvus import MilvusClient, DataType# 2. 连接到 Milvus 服务器：uri 为服务器地址（本地默认 19530 端口）
client = MilvusClient(uri="http://localhost:19530")  # 生产环境需替换为实际服务器地址# 3. 创建 Schema：auto_id=False 表示手动指定主键，enable_dynamic_field=True 启用动态字段（可选）
schema = client.create_schema(auto_id=False, enable_dynamic_field=True)# 4. 添加主键字段：product_id 作为唯一标识，类型为 INT64，设为必填主键
schema.add_field(field_name="product_id",datatype=DataType.INT64,is_primary=True  # 标记为主键字段
)# 5. 添加向量字段：用于向量搜索，类型为 FLOAT_VECTOR，维度 dim=5（需与实际向量维度一致）
schema.add_field(field_name="vector",datatype=DataType.FLOAT_VECTOR,dim=5  # 向量维度，根据业务数据调整（如图片特征向量可能为 512 维）
)# 6. 定义 JSON 字段：字段名为 metadata，类型为 JSON，nullable=True 允许空值（可选配置）
schema.add_field(field_name="metadata",datatype=DataType.JSON,nullable=True  # 设为 True 时，该字段可插入空值（如部分产品无供应商信息）
)# 7. 创建集合：集合名 product_catalog，关联上述定义的 Schema
client.create_collection(collection_name="product_catalog",  # 集合名称（需唯一）schema=schema  # 绑定之前创建的 Schema
)print("集合创建成功!")  # 执行成功后输出提示

关键说明：JSON 字段必须在 Schema 中明确声明，可通过 nullable 控制是否允许空值，需与主键、向量字段配合使用（Milvus 集合需包含向量字段用于向量搜索）。

3.2 插入数据

# 1. 准备插入的数据：entities 为列表，每个元素是一条完整记录（包含主键、向量、JSON 字段）
entities = [{"product_id": 1,  # 主键值（需唯一，auto_id=False 时必须手动指定）"vector": [0.1, 0.2, 0.3, 0.4, 0.5],  # 向量数据（维度需与 Schema 中 dim 一致）"metadata": {  # JSON 字段内容，结构与 Schema 定义一致"category": "electronics","brand": "BrandA","in_stock": True,"price": 99.99,"string_price": "99.99","tags": ["clearance", "summer_sale"],  # 数组类型值"supplier": {  # 嵌套对象类型值"name": "SupplierX","country": "USA","contact": {  # 三级嵌套"email": "support@supplierx.com","phone": "+1-800-555-0199"}}}},{"product_id": 2,  # 第二条记录主键（需与第一条不同）"vector": [0.2, 0.3, 0.4, 0.5, 0.6],"metadata": {"category": "clothing","brand": "BrandB","in_stock": False,"price": 49.99,"tags": ["new", "winter"],"supplier": {"name": "SupplierY","country": "China","contact": {"email": "contact@suppliery.com","phone": "+86-10-555-0199"}}}}
]# 2. 插入数据：指定集合名和待插入数据，返回结果包含插入成功的 ID
result = client.insert(collection_name="product_catalog",  # 目标集合名data=entities  # 待插入的记录列表
)# 3. 输出插入结果：result['ids'] 包含所有插入成功的主键值
print(f"插入成功!插入的ID: {result['ids']}")

关键说明：JSON 字段的结构可灵活调整（如第二条记录缺少 string_price 字段），但相同键的数据类型建议保持一致（避免同一键既存数值又存字符串）；向量维度必须与 Schema 定义的 dim 匹配，否则插入失败。

3.3 基本查询操作

3.3.1 使用 JSON 路径语法进行筛选

# 1. 定义过滤条件：筛选 metadata 中 category 为 "electronics" 的记录
# 语法规则：JSON 字段名["键名"]，字符串值需用双引号包裹
filter_expr = 'metadata["category"] == "electronics"'# 2. 执行搜索：结合向量搜索和属性过滤
results = client.search(collection_name="product_catalog",  # 目标集合名data=[[0.1, 0.2, 0.3, 0.4, 0.5]],  # 查询向量（与待搜索向量维度一致）limit=5,  # 返回结果的最大数量filter=filter_expr,  # 应用上述过滤条件output_fields=["product_id", "metadata"]  # 指定返回的字段（主键 + JSON 字段）
)# 3. 解析并输出结果
print("电子产品搜索结果:")
for result in results:  # 遍历返回的结果列表# result['entity'] 包含单条记录的字段数据print(f"产品ID: {result['entity']['product_id']}, 距离: {result['distance']}")print(f"元数据: {result['entity']['metadata']}")

语法说明：metadata["category"] 是 JSON 路径表达式，用于定位 JSON 字段中的具体键；distance 是查询向量与结果向量的相似度距离（值越小越相似）。

3.3.2 过滤嵌套键

# 1. 定义过滤条件：筛选嵌套键 supplier -> country 为 "USA" 的记录
# 嵌套路径语法：JSON 字段名["父键"]["子键"]（支持多层嵌套）
filter_expr = 'metadata["supplier"]["country"] == "USA"'# 2. 执行搜索：查询向量与上例一致，仅过滤条件不同
results = client.search(collection_name="product_catalog",data=[[0.1, 0.2, 0.3, 0.4, 0.5]],limit=5,filter=filter_expr,output_fields=["product_id", "metadata"]
)# 3. 输出结果
print("美国供应商产品搜索结果:")
for result in results:print(f"产品ID: {result['entity']['product_id']}")

关键说明：嵌套路径需准确匹配 JSON 结构，若某条记录缺少路径中的某个键（如无 supplier 字段），则该记录不会被匹配到。

3.4 使用 JSON 特定操作符

3.4.1 数组包含匹配（json_contains）

# 1. 过滤条件：查找 tags 数组中包含 "summer_sale" 的产品
# json_contains(数组路径, 目标值)：判断数组是否包含指定值
filter_expr = 'json_contains(metadata["tags"], "summer_sale")'results = client.search(collection_name="product_catalog",data=[[0.1, 0.2, 0.3, 0.4, 0.5]],limit=5,filter=filter_expr,output_fields=["product_id", "metadata"]
)print("夏季促销产品:")
for result in results:print(f"产品ID: {result['entity']['product_id']}")

3.4.2 数组多值包含匹配（json_contains_any）

# 1. 过滤条件：查找 tags 数组中包含 "electronics"、"new" 或 "clearance" 任意一个值的产品
# json_contains_any(数组路径, [值1, 值2, ...])：判断数组是否包含任意一个目标值
filter_expr = 'json_contains_any(metadata["tags"], ["electronics", "new", "clearance"])'results = client.search(collection_name="product_catalog",data=[[0.1, 0.2, 0.3, 0.4, 0.5]],limit=5,filter=filter_expr,output_fields=["product_id", "metadata"]
)print("包含指定标签的产品:")
for result in results:print(f"产品ID: {result['entity']['product_id']}, 标签: {result['entity']['metadata']['tags']}")

操作符说明：json_contains 适用于精确匹配单个值，json_contains_any 适用于多值 “或” 匹配，两者均专门针对 JSON 数组字段设计。

四、JSON 索引

4.1 JSON 索引概述

JSON 索引是针对 JSON 字段中特定路径的键创建的索引，用于加速等价查询（如 ==）和范围查询（如 >=、<=）。它不直接索引整个 JSON 对象，而是聚焦于用户指定的关键路径，适用于以下场景：

JSON 数据结构固定，核心查询键明确（如频繁按 category 或 price 查询）。
需要对嵌套键（如 supplier.contact.email）进行高效查询。
需对特定数据类型（如字符串、数值）进行精确匹配或范围筛选。
希望在保持 JSON 灵活性的同时，获得接近传统列索引的查询性能。

4.2 创建 JSON 索引

# 1. 准备索引参数：通过 prepare_index_params() 初始化索引配置对象
index_params = client.prepare_index_params()# 示例1：在 JSON 一级字段 category 上创建索引
index_params.add_index(field_name="metadata",  # 索引对应的 JSON 字段名（必须是 Schema 中声明的 JSON 字段）index_type="AUTOINDEX",  # 索引类型（Milvus 自动选择最优索引算法）index_name="category_index",  # 索引名称（自定义，需唯一）params={"json_path": 'metadata["category"]',  # 索引的 JSON 路径（一级字段）"json_cast_type": "varchar"  # 索引数据类型（category 是字符串，对应 varchar）}
)# 示例2：在嵌套字段 supplier.contact.email 上创建索引
index_params.add_index(field_name="metadata",index_type="AUTOINDEX",index_name="email_index",params={"json_path": 'metadata["supplier"]["contact"]["email"]',  # 三级嵌套路径"json_cast_type": "varchar"  # email 是字符串类型}
)# 示例3：索引时转换数据类型（字符串转数值）
index_params.add_index(field_name="metadata",index_type="AUTOINDEX",index_name="string_to_double_index",params={"json_path": 'metadata["string_price"]',  # 原始字段是字符串类型（"99.99"）"json_cast_type": "double",  # 目标索引类型（双精度数值）"json_cast_function": "STRING_TO_DOUBLE"  # 类型转换函数（字符串转双精度）}
)# 示例4：索引整个 JSON 对象
index_params.add_index(field_name="metadata",index_type="AUTOINDEX",index_name="metadata_full_index",params={"json_path": "metadata",  # 路径直接指定整个 JSON 字段"json_cast_type": "JSON"  # 索引类型为 JSON（适用于整体匹配场景）}
)# 2. 应用索引配置：将索引创建到目标集合
client.create_index(collection_name="product_catalog",index_params=index_params
)print("JSON 索引创建成功!")

关键说明：

索引路径必须是 JSON 字段中实际存在的键（可嵌套），否则索引创建无效。
json_cast_type 需与目标键的数据类型匹配（如字符串用 varchar，数值用 double）。
类型转换函数仅支持 Milvus 内置函数（如 STRING_TO_DOUBLE），需确保转换后的数据格式合法（如不能将非数字字符串转为 double）。

4.3 支持的数据类型

数据类型	描述	示例 JSON 值	适用场景
BOOL / bool	布尔值类型索引	true, false	筛选布尔型属性（如库存状态 in_stock ）
DOUBLE / double	数值类型索引（含整数、浮点数）	42, 99.99	价格、数量等数值的等价 / 范围查询
VARCHAR / varchar	字符串类型索引	"electronics", "BrandA"	类别、品牌、名称等字符串的精确匹配
ARRAY_BOOL / array_bool	布尔值数组索引	[true, false, true]	布尔数组的包含匹配（如多状态标记）
ARRAY_DOUBLE / array_double	数值数组索引	[1.2, 3.14, 42]	数值数组的包含 / 范围匹配（如多规格尺寸）
ARRAY_VARCHAR / array_varchar	字符串数组索引	["tag1", "tag2", "tag3"]	标签、关键词等数组的包含匹配
JSON / json	整个 JSON 对象 / 子对象索引	任意 JSON 对象	整体 JSON 结构的匹配查询（较少用）

五、JSON 切碎 (JSON Shredding)

5.1 JSON 切碎概述

JSON 切碎是 Milvus 提供的 JSON 存储优化技术，核心是将传统的行式存储的 JSON 数据，转换为优化的列式存储。简单来说，就是把 JSON 中的每个键（包括嵌套键）拆分为独立的 “虚拟列”，按列存储数据。这样做的优势是：

查询时仅需扫描目标键对应的列，无需加载整个 JSON 对象，提升查询速度。
保持 JSON 数据模型的灵活性（仍可动态扩展字段），同时获得列式存储的性能优势。
适用于 JSON 键较多、查询频繁且分散的场景（如多条件组合查询）。

5.2 启用 JSON 切碎

JSON 切碎需通过修改 Milvus 配置文件 milvus.yaml 启用，配置步骤如下：

找到 Milvus 安装目录下的 conf/milvus.yaml 文件（默认路径）。
找到 common 配置段，添加或修改以下参数：

common:enabledJSONKeyStats: true  # 核心开关：启用 JSON 键统计构建和加载流程（必须设为 true）

保存配置文件，重启 Milvus 服务使配置生效。

5.3 配置参数

参数名称	说明	默认值	调整建议
common.enabledJSONKeyStats	控制是否启用 JSON 切碎功能	false	必须设为 true 才能激活切碎，否则后续配置无效
common.usingJsonStatsForQuery	控制查询时是否使用切碎后的数据加速	true	正常场景保持 true；若查询失败，可设为 false 恢复原始查询路径
queryNode.mmap.jsonStats	决定是否使用内存映射（mmap）加载切碎数据	true	mmap 可减少内存占用，提升加载速度，建议保持默认
dataCoord.jsonStatsMaxShreddingColumns	切碎后允许的最大虚拟列数量（即 JSON 键的最大数量）	1024	若 JSON 包含数千个频繁使用的键，需增大该值（如 2048），避免部分键无法切碎
dataCoord.jsonStatsShreddingRatioThreshold	JSON 键被切碎的最小出现率（比例）	0.3	出现率 = 包含该键的记录数 / 总记录数；降低阈值（如 0.1）可切碎更多低频率键，提升小众键查询效率

5.3 完整配置示例

common:enabledJSONKeyStats: true  # 启用 JSON 切碎usingJsonStatsForQuery: true  # 使用切碎数据加速查询
dataCoord:jsonStatsMaxShreddingColumns: 1024  # 最大切碎列数（根据实际场景调整）jsonStatsShreddingRatioThreshold: 0.3  # 最小出现率（30%）
queryNode:mmap:jsonStats: true  # 启用 mmap 加载切碎数据

六、查询优化技术比较

技术	最适合场景	数组加速	优势	局限性
JSON 索引	少数频繁查询的键、特定数组键查询	是（仅在索引的数组键上）	针对性强，查询速度快，配置灵活	需预先选择索引键，Schema 变化后需重新维护索引
JSON 切碎	复杂 JSON 文档、多个分散查询键、多条件组合查询	否（不加速数组内的值）	无需手动配置索引，自动适配多键查询，保持 JSON 灵活性	额外占用少量存储，数组查询无加速效果
NGRAM 索引	通配符搜索、文本字段子串匹配（如模糊查询）	不适用	支持灵活的文本模糊查询（如 %elect% ）	不适用于数字、布尔值，无法进行范围筛选

优化组合建议

常规场景：JSON 切碎 + 高频键 JSON 索引，兼顾多键查询灵活性和高频查询效率。
文本搜索场景：JSON 切碎 + NGRAM 索引，满足结构化查询和文本模糊查询需求。
数组查询场景：JSON 索引（针对数组键），提升数组包含匹配的速度。

七、应用场景示例

7.1 电商产品搜索（多条件组合查询）

def search_products_by_filters(client, query_vector, category=None, brand=None, min_price=None, max_price=None, tags=None
):"""根据多种条件搜索电商产品（支持类别、品牌、价格范围、标签筛选）参数说明：- client: MilvusClient 实例（已连接服务器）- query_vector: 查询向量（维度需与集合向量字段一致）- category: 产品类别（可选，如 "electronics"）- brand: 品牌（可选，如 "BrandA"）- min_price: 最低价格（可选）- max_price: 最高价格（可选）- tags: 标签列表（可选，如 ["clearance", "new"]）返回：满足条件的搜索结果列表"""filters = []  # 存储所有过滤条件（最终拼接为逻辑表达式）# 1. 类别筛选（字符串精确匹配）if category:filters.append(f'metadata["category"] == "{category}"')# 2. 品牌筛选（字符串精确匹配）if brand:filters.append(f'metadata["brand"] == "{brand}"')# 3. 价格范围筛选（数值比较）if min_price is not None and max_price is not None:# 同时指定最低和最高价格：价格 >= min_price 且 <= max_pricefilters.append(f'metadata["price"] >= {min_price} && metadata["price"] <= {max_price}')elif min_price is not None:# 仅指定最低价格：价格 >= min_pricefilters.append(f'metadata["price"] >= {min_price}')elif max_price is not None:# 仅指定最高价格：价格 <= max_pricefilters.append(f'metadata["price"] <= {max_price}')# 4. 标签筛选（数组包含任意一个标签）if tags:# 生成每个标签的包含条件（json_contains），用 || 连接（逻辑或）tag_filters = [f'json_contains(metadata["tags"], "{tag}")' for tag in tags]filters.append(f'({" || ".join(tag_filters)})')# 拼接所有过滤条件（用 && 连接，逻辑与）filter_expr = " && ".join(filters) if filters else ""# 执行搜索results = client.search(collection_name="product_catalog",  # 目标集合data=[query_vector],  # 查询向量limit=10,  # 返回最多 10 条结果filter=filter_expr if filter_expr else None,  # 过滤条件（无条件时设为 None）output_fields=["product_id", "metadata"]  # 返回字段)return results# 使用示例：搜索电子类、价格 50-200 元、包含 "clearance" 或 "new" 标签的产品
if __name__ == "__main__":# 1. 重新连接 Milvus 服务器（若之前断开连接）client = MilvusClient(uri="http://localhost:19530")# 2. 定义查询向量（示例向量，实际应使用产品特征向量）query_vector = [0.15, 0.25, 0.35, 0.45, 0.55]# 3. 调用函数执行搜索results = search_products_by_filters(client=client,query_vector=query_vector,category="electronics",  # 类别：电子类min_price=50,  # 最低价格 50 元max_price=200,  # 最高价格 200 元tags=["clearance", "new"]  # 标签：清仓或新品)# 4. 输出结果print("筛选结果:")for result in results:product = result['entity']print(f"ID: {product['product_id']}, 价格: {product['metadata']['price']}, 标签: {product['metadata']['tags']}")

7.2 内容管理系统（文档搜索）

def search_documents(client, query_vector, document_type=None, author=None, publish_date=None, keywords=None
):"""在内容管理系统中搜索文档（支持文档类型、作者、发布日期、关键词筛选）参数说明：- client: MilvusClient 实例- query_vector: 文档特征向量- document_type: 文档类型（可选，如 "article"）- author: 作者（可选）- publish_date: 发布日期（可选，如 "2024-01-01"，支持 >= 筛选）- keywords: 关键词列表（可选，如 ["AI", "Milvus"]）返回：满足条件的文档列表"""filters = []# 1. 文档类型筛选if document_type:filters.append(f'metadata["document_type"] == "{document_type}"')# 2. 作者筛选if author:filters.append(f'metadata["author"] == "{author}"')# 3. 发布日期筛选（日期字符串比较，需确保存储格式一致）if publish_date:filters.append(f'metadata["publish_date"] >= "{publish_date}"')# 4. 关键词筛选（数组包含匹配）if keywords:keyword_filters = [f'json_contains(metadata["keywords"], "{keyword}")' for keyword in keywords]filters.append(f'({" || ".join(keyword_filters)})')# 拼接过滤条件filter_expr = " && ".join(filters) if filters else ""# 执行搜索results = client.search(collection_name="document_store",  # 文档集合名（需提前创建）data=[query_vector],limit=10,filter=filter_expr if filter_expr else None,output_fields=["doc_id", "metadata", "content_summary"]  # 返回文档ID、元数据、内容摘要)return results# 调用示例（搜索 2024 年之后发布的 AI 相关文章）
query_vector = [0.3, 0.4, 0.5, 0.6, 0.7]  # 文档特征向量
results = search_documents(client=client,query_vector=query_vector,document_type="article",publish_date="2024-01-01",keywords=["AI", "Milvus"]
)# 输出结果
for result in results:doc = result['entity']print(f"文档ID: {doc['doc_id']}, 作者: {doc['metadata']['author']}, 发布日期: {doc['metadata']['publish_date']}")