Clustering Compaction

Clustering Compaction 旨在提高搜索性能，降低大型 Collection 的成本。本指南将帮助您了解 Clustering Compaction 以及该功能如何提高搜索性能。

概述

Milvus 将输入的实体存储在 Collection 中的 Segment 中，并在 Segment 已满时将其封存。如果出现这种情况，就会创建一个新的 Segment 来容纳更多的实体。因此，实体会任意地分布在不同的 Segment 中。这种分布要求 Milvus 搜索多个 Segment，以找到与给定查询向量最近的邻居。

如果 Milvus 可以根据特定字段中的值将实体分布在不同的 Segment 中，那么搜索范围就可以限制在一个 Segment 内，从而提高搜索性能。

Clustering Compaction 是 Milvus 的一项功能，它能根据标量字段中的值在 Collection 的 Segment 之间重新分配 Entity。要启用此功能，首先需要选择一个标量字段作为 Clustering Key。这样，当实体的 Clustering Key 值在特定范围内时，Milvus 就能将实体重新分配到 Segment 中。当你触发 Clustering Compaction 时，Milvus 会生成/更新一个名为 PartitionStats 的全局索引，它记录了 Segment 与 Clustering Key 值之间的映射关系。

以 PartitionStats 为参考，Milvus 可以在收到带有 Clustering Key 值的搜索/查询请求时，剪切不相关的数据，并将搜索范围限制在与键值映射的 Segment 内，从而提高搜索性能。有关性能改进的详细信息，请参阅 基准测试。

使用 Clustering Compaction

Milvus 的 Clustering Compaction 功能具有高度可配置性。你可以选择手动触发，也可以将其设置为由 Milvus 每隔一段时间自动触发。要启用 Clustering Compaction，请执行以下操作：

全局配置

您需要修改 Milvus 配置文件，如下所示。

YAML

dataCoord:
  compaction:
    clustering:
      enable: true
      autoEnable: false
      triggerInterval: 600
      minInterval: 3600
      maxInterval: 259200
      newDataSizeThreshold: 512m
      timeout: 7200

queryNode:
  enableSegmentPrune: true

datanode:
  clusteringCompaction:
    memoryBufferRatio: 0.1
    workPoolSize: 8
common:
  usePartitionKeyAsClusteringKey: true

配置项目	说明	默认值
dataCoord.compaction.clustering
enable	指定是否启用 Clustering Compaction。如果需要为每个具有聚类密钥的 Collection 启用此功能，请将其设置为 true。	false
autoEnable	指定是否启用自动触发 Clustering Compaction。将此设置为 true 表示 Milvus 在指定的时间间隔对具有 Clustering Key 的 Collection 进行压缩。	false
triggerInterval	以毫秒为单位指定 Milvus 开始 Clustering Compaction 的时间间隔。只有将 autoEnable 设置为 true 时才适用。
minInterval	以毫秒为单位指定最小间隔。仅当设置 autoEnable 至 true 时适用。 将其设置为大于 triggerInterval 的整数有助于避免在短时间内重复压缩。
maxInterval	以毫秒为单位指定最大间隔。只有将 autoEnable 设置为 true 时才适用。 一旦 Milvus 检测到某个 Collection 的 Clustering Compaction 持续时间超过此值，就会强制进行 Clustering Compaction。
newDataSizeThreshold	指定触发 Clustering Compaction 的上阈值。这仅适用于将 autoEnable 设置为 true 时。 一旦 Milvus 检测到 Collection 中的数据量超过此值，就会启动 Clustering Compaction 进程。
timeout	指定 Clustering Compaction 的超时持续时间。如果执行时间超过此值，则 Clustering Compaction 失败。
queryNode
enableSegmentPrune	指定 Milvus 是否在收到搜索/查询请求时参考 PartitionStats 来剪切数据。将此值设为 true，这样 Milvus 就能在收到搜索/查询请求时通过引用 PartitionStats 来修剪数据。
dataNode.clusteringCompaction
memoryBufferRatio	指定集群压缩任务的内存缓冲区比率。 当数据大小超过使用此比率计算出的分配缓冲区大小时，Milvus 会刷新数据。
workPoolSize	指定 Clustering Compaction 任务的工作池大小。
common
usePartitionKeyAsClusteringKey	指定是否使用 Collection 中的 Partition Key 作为聚类密钥。将此设置为 "true"，Milvus 就会把 Collection 中的 Partition Key 作为聚类密钥。 你可以在 Collection 中通过显式设置聚类密钥来覆盖此设置。

要将上述更改应用到 Milvus 集群，请按照 使用 Helm 配置 Milvus 和 使用 Milvus Operator 配置 Milvus 中的步骤操作。

Collection 配置

要在特定 Collection 中进行 Clustering Compaction，应从 Collection 中选择一个标量字段作为 Clustering Key。

Python

Python

from pymilvus import MilvusClient, DataType

CLUSTER_ENDPOINT="http://localhost:19530"
TOKEN="root:Milvus"

client = MilvusClient(
    uri=CLUSTER_ENDPOINT,
    token=TOKEN
)

schema = MilvusClient.create_schema()
schema.add_field("id", DataType.INT64, is_primary=True, auto_id=False)
schema.add_field("key", DataType.INT64, is_clustering_key=True)
schema.add_field("var", DataType.VARCHAR, max_length=1000)
schema.add_field("vector", DataType.FLOAT_VECTOR, dim=5)

client.create_collection(
    collection_name="clustering_test",
    schema=schema
)

Java

Java

import io.milvus.v2.client.ConnectConfig;
import io.milvus.v2.client.MilvusClientV2;
import io.milvus.v2.common.DataType;
import io.milvus.v2.service.collection.request.AddFieldReq;
import io.milvus.v2.service.collection.request.CreateCollectionReq;

MilvusClientV2 client = new MilvusClientV2(ConnectConfig.builder()
        .uri("http://localhost:19530")
        .token("root:Milvus")
        .build());

CreateCollectionReq.CollectionSchema schema = client.createSchema();

schema.addField(AddFieldReq.builder()
        .fieldName("id")
        .dataType(DataType.Int64)
        .isPrimaryKey(true)
        .autoID(false)
        .build());

schema.addField(AddFieldReq.builder()
        .fieldName("key")
        .dataType(DataType.Int64)
        .isClusteringKey(true)
        .build());

schema.addField(AddFieldReq.builder()
        .fieldName("var")
        .dataType(DataType.VarChar)
        .maxLength(1000)
        .build());

schema.addField(AddFieldReq.builder()
        .fieldName("vector")
        .dataType(DataType.FloatVector)
        .dimension(5)
        .build());

CreateCollectionReq requestCreate = CreateCollectionReq.builder()
        .collectionName("clustering_test")
        .collectionSchema(schema)
        .build();
client.createCollection(requestCreate);

Node.js

JavaScript

import { MilvusClient, DataType } from '@zilliz/milvus2-sdk-node';

const CLUSTER_ENDPOINT = 'http://localhost:19530';
const TOKEN = 'root:Milvus';
const client = new MilvusClient({
  address: CLUSTER_ENDPOINT,
  token: TOKEN,
});
const schema = [
    {
      name: 'id',
      type: DataType.Int64,
      is_primary_key: true,
      autoID: false,
    },
    {
      name: 'key',
      type: DataType.Int64,
      is_clustering_key: true,
    },
    {
      name: 'var',
      type: DataType.VarChar,
      max_length: 1000,
      is_primary_key: false,
    },
    {
      name: 'vector',
      type: DataType.FloatVector,
      dim: 5,
    },
  ];

  await client.createCollection({
    collection_name: 'clustering_test',
    schema: schema,
  });

Go

Go

schema := entity.NewSchema().
    WithField(entity.NewField().WithName("id").WithDataType(entity.FieldTypeInt64).WithIsPrimaryKey(true)).
    WithField(entity.NewField().WithName("key").WithDataType(entity.FieldTypeInt64).WithIsClusteringKey(true)).
    WithField(entity.NewField().WithName("var").WithDataType(entity.FieldTypeVarChar).WithMaxLength(1000)).
    WithField(entity.NewField().WithName("vector").WithDataType(entity.FieldTypeFloatVector).WithDim(5))
if err := client.CreateCollection(ctx, milvusclient.NewCreateCollectionOption("clustering_test", schema)); err != nil {
    log.Fatal(err)
}

cURL

Shell

curl --request POST \
  --url "http://localhost:19530/v2/vectordb/collections/create" \
  --header "Authorization: Bearer root:Milvus" \
  --header "Content-Type: application/json" \
  --data '{
    "collectionName": "clustering_test",
    "schema": {
      "autoID": false,
      "fields": [
        {"fieldName": "id", "dataType": "Int64", "isPrimary": true},
        {"fieldName": "key", "dataType": "Int64", "isClusteringKey": true},
        {"fieldName": "var", "dataType": "VarChar", "elementTypeParams": {"max_length": 1000}},
        {"fieldName": "vector", "dataType": "FloatVector", "elementTypeParams": {"dim": 5}}
      ]
    }
  }'

说明

您可以使用以下数据类型的标量字段作为 Clustering Key：Int8,Int16,Int32,Int64,Float,Double 和 VarChar。

触发 Clustering Compaction

如果启用了自动 Clustering Compaction，Milvus 会在指定的时间间隔自动触发 Clustering Compaction。或者，您也可以按如下方式手动触发 Clustering Compaction：

Python

Python

# trigger a manual compaction
job_id = client.compact(
    collection_name="clustering_test",
    is_clustering=True
)

# get the compaction state
client.get_compaction_state(
    job_id=job_id,
)

Java

Java

import io.milvus.v2.service.utility.request.CompactReq;
import io.milvus.v2.service.utility.request.GetCompactionStateReq;
import io.milvus.v2.service.utility.response.CompactResp;
import io.milvus.v2.service.utility.response.GetCompactionStateResp;

CompactResp compactResp = client.compact(CompactReq.builder()
        .collectionName("clustering_test")
        .isClustering(true)
        .build());

GetCompactionStateResp stateResp = client.getCompactionState(GetCompactionStateReq.builder()
        .compactionID(compactResp.getCompactionID())
        .build());

System.out.println(stateResp.getState());

Node.js

JavaScript

// trigger a manual compaction
const {compactionID} = await client.compact({
    collection_name: "clustering_test",
    is_clustering: true
});

// get the compaction state
await client.getCompactionState({
    compactionID: compactionID,
});

Go

当前 Go Client v2 的 CompactOption 只接受 Collection 名称，未公开 is_clustering 参数，无法保证触发的是 Clustering Compaction。

cURL

Shell

curl --request POST \
  --url "http://localhost:19530/v2/vectordb/collections/compact" \
  --header "Authorization: Bearer root:Milvus" \
  --header "Content-Type: application/json" \
  --data '{"collectionName": "clustering_test", "isClustering": true}'

# REST v2 OpenAPI currently documents an empty compact response.
# The state endpoint can be called after a compaction job ID is available.
JOB_ID="<compaction-job-id>"
curl --request POST \
  --url "http://localhost:19530/v2/vectordb/collections/get_compaction_state" \
  --header "Authorization: Bearer root:Milvus" \
  --header "Content-Type: application/json" \
  --data "{\"jobID\": \"${JOB_ID}\"}"

基准测试

数据量和查询模式共同决定了 Clustering Compaction 所能带来的性能提升。一项内部基准测试表明，Clustering Compaction 最多可将每秒查询次数（QPS）提高 25 倍。

该基准测试是在一个包含来自 2000 万个 768 维 LAION 数据集的实体的 Collection 上进行的，该数据集的 key 字段被指定为聚类密钥。在 Collection 中触发 Clustering Compaction 后，会发送并发搜索，直到 CPU 使用率达到高水位。

搜索过滤器	剪切率	延迟					Reqs/s
		平均值	最小值	最大值	中位数	TP99
N/A	0%	1685	672	2294	1710	2291	17.75
密钥>200 和密钥 < 800	40.2%	1045	47	1828	1085	1617	28.38
键>200 和键 < 600	59.8%	829	45	1483	882	1303	35.78
键>200 和键 < 400	79.5%	550	100	985	584	898	54.00
键==1000	99%	68	24	1273	70	246	431.41

随着搜索筛选器中搜索范围的缩小，剪切率也随之增加。这意味着在搜索过程中会跳过更多的实体。比较第一行和最后一行的统计数据，可以发现不进行 Clustering Compaction 的搜索需要扫描整个 Collection。另一方面，使用特定键进行 Clustering Compaction 的搜索可以实现高达 25 倍的改进。

最佳实践

以下是一些有效使用 Clustering Compaction 的提示：

• 为数据量较大的 Collection 启用此功能。

Collection 中的数据量越大，搜索性能就越高。对于超过 100 万个实体的 Collection，启用此功能是一个不错的选择。

• 选择合适的聚类关键字。

可以使用通常用作筛选条件的标量字段作为聚类关键字。对于保存多个租户数据的 Collection，可以利用区分一个租户和另一个租户的字段作为聚类密钥。

• 使用 Partition Key 作为聚类密钥。

如果你想为 Milvus 实例中的所有 Collection 启用此功能，或者在使用 Partition Key 的大型 Collection 中仍面临性能问题，可以将 common.usePartitionKeyAsClusteringKey 设置为 true。通过这样做，当你选择 Collection 中的标量字段作为 Partition Key 时，你将拥有一个 Clustering Key 和一个 Partition Key。

请注意，此设置不会阻止您选择另一个标量字段作为 Clustering Key。明确指定的 Clustering Key 始终优先。