版本：v2.5.x

使用 Milvus Operator 配置消息存储

Milvus 使用 RocksMQ、Pulsar 或 Kafka 管理最近更改的日志、输出流日志并提供日志订阅。本主题介绍如何在使用 Milvus Operator 安装 Milvus 时配置消息存储依赖关系。有关详细信息，请参阅 Milvus Operator 存储库中的使用 Milvus Operator 配置消息存储。

本主题假定您已部署 Milvus Operator。

说明

有关详细信息，请参阅部署 Milvus Operator。

您需要指定使用 Milvus Operator 启动 Milvus 群集的配置文件。

kubectl apply -f https://raw.githubusercontent.com/zilliztech/milvus-operator/main/config/samples/milvus_cluster_default.yaml

您只需编辑milvus_cluster_default.yaml 中的代码模板，即可配置第三方依赖关系。以下各节将分别介绍如何配置对象存储、etcd 和 Pulsar。

开始之前

下表显示了 Milvus 独立模式和集群模式是否支持 RocksMQ、NATS、Pulsar 和 Kafka。

	RocksMQ	NATS	脉冲星	卡夫卡
独立模式	✔️	✔️	✔️	✔️
集群模式	✖️	✖️	✔️	✔️

指定消息存储还有其他限制：

一个 Milvus 实例只支持一个消息存储。不过，我们仍然向后兼容为一个实例设置多个消息存储空间。优先级如下：
- 独立模式： RocksMQ（默认） > Pulsar > Kafka
- 集群模式：脉冲星（默认） > 卡夫卡
- 为了向后兼容，2.3 中引入的 Nats 不参与这些优先级规则。
Milvus 系统运行时不能更改消息存储。
仅支持 Kafka 2.x 或 3.x 版本。

配置 RocksMQ

RocksMQ 是 Milvus Standalone 的默认消息存储。

说明

目前，你只能通过 Milvus Operator 配置 RocksMQ 作为 Milvus Standalone 的消息存储。

示例

下面的示例配置了一个 RocksMQ 服务。

apiVersion: milvus.io/v1beta1
kind: Milvus
metadata:
  name: milvus
spec:
  mode: standalone
  dependencies:
    msgStreamType: rocksmq
    rocksmq:
      persistence:
        enabled: true
        pvcDeletion: true
        persistentVolumeClaim:
          spec:
            accessModes: ["ReadWriteOnce"]
            storageClassName: "local-path"  # Specify your storage class
            resources:
              requests:
                storage: 10Gi  # Specify your desired storage size
  components: {}
  config: {}

主要配置选项：

msgStreamTyperocksmq: 明确设置 RocksMQ 为消息队列。
persistence.enabled:启用 RocksMQ 数据的持久存储
persistence.pvcDeletion:为 true 时，PVC 将在 Milvus 实例删除时被删除。
persistentVolumeClaim.spec:标准 Kubernetes PVC 规范
accessModes:通常ReadWriteOnce 用于块存储
storageClassName:您的集群的存储类
storage:持久卷的大小

已废弃：NATS (natsmq) 已被弃用，并将在 Milvus v2.6 中移除。请使用 RocksMQ、Pulsar 或 Kafka 代替。

配置 NATS

NATS 是 NATS 的另一种消息存储方式。

示例

下面的示例配置了 NATS 服务。

apiVersion: milvus.io/v1alpha1
kind: Milvus
metadata:
  name: milvus
spec:
  dependencies:
    msgStreamType: 'natsmq'
    natsmq:
      # server side configuration for natsmq.
      server:
        # 4222 by default, Port for nats server listening.
        port: 4222
        # /var/lib/milvus/nats by default, directory to use for JetStream storage of nats.
        storeDir: /var/lib/milvus/nats
        # (B) 16GB by default, Maximum size of the 'file' storage.
        maxFileStore: 17179869184
        # (B) 8MB by default, Maximum number of bytes in a message payload.
        maxPayload: 8388608
        # (B) 64MB by default, Maximum number of bytes buffered for a connection applies to client connections.
        maxPending: 67108864
        # (√ms) 4s by default, waiting for initialization of natsmq finished.
        initializeTimeout: 4000
        monitor:
          # false by default, If true enable debug log messages.
          debug: false
          # true by default, If set to false, log without timestamps.
          logTime: true
          # no log file by default, Log file path relative to.. .
          logFile:
          # (B) 0, unlimited by default, Size in bytes after the log file rolls over to a new one.
          logSizeLimit: 0
        retention:
          # (min) 3 days by default, Maximum age of any message in the P-channel.
          maxAge: 4320
          # (B) None by default, How many bytes the single P-channel may contain. Removing oldest messages if the P-channel exceeds this size.
          maxBytes:
          # None by default, How many message the single P-channel may contain. Removing oldest messages if the P-channel exceeds this limit.
          maxMsgs:
  components: {}
  config: {}

要将消息存储从 RocksMQ 迁移到 NATS，请执行以下操作：

停止所有 DDL 操作符。
调用 FlushAll API，然后在 API 调用执行完毕后停止 Milvus。
将msgStreamType 更改为natsmq ，并对spec.dependencies.natsmq 中的 NATS 设置进行必要更改。
再次启动 Milvus 并检查是否：
- 日志中是否存在读作mqType=natsmq 的日志条目。
- spec.dependencies.natsmq.server.storeDir 中指定的目录中是否有名为jetstream 的目录。
(可选）备份并清理 RocksMQ 存储目录中的数据文件。

说明

在 RocksMQ 和 NATS 之间做选择？

RockMQ 使用 CGO 与 RocksDB 交互并自行管理内存，而 Milvus 安装中嵌入的纯 Go NATS 则将内存管理委托给 Go 的垃圾收集器（GC）。

在数据包小于 64 kb 的情况下，RocksDB 在内存使用率、CPU 使用率和响应时间方面都更胜一筹。另一方面，如果数据包大于 64 kb，NATS 则会在内存充足和 GC 调度理想的情况下，在响应时间方面表现出色。

目前，建议您仅在实验中使用 NATS。

配置 Pulsar

Pulsar 管理最近更改的日志、输出流日志并提供日志订阅。Milvus Standalone 和 Milvus 集群都支持为消息存储配置 Pulsar。不过，使用 Milvus Operator，只能将 Pulsar 配置为 Milvus 集群的消息存储。添加spec.dependencies.pulsar 下的必填字段以配置 Pulsar。

pulsar 支持external 和inCluster 。

外部 Pulsar

external 表示使用外部 Pulsar 服务。用于配置外部 Pulsar 服务的字段包括：

external: true 值表示 Milvus 使用外部 Pulsar 服务。
endpoints:Pulsar 的端点。

示例

下面的示例配置了外部 Pulsar 服务。

apiVersion: milvus.io/v1alpha1
kind: Milvus
metadata:
  name: my-release
  labels:
    app: milvus
spec:
  dependencies: # Optional
    pulsar: # Optional
      # Whether (=true) to use an existed external pulsar as specified in the field endpoints or
      # (=false) create a new pulsar inside the same kubernetes cluster for milvus.
      external: true # Optional default=false
      # The external pulsar endpoints if external=true
      endpoints:
      - 192.168.1.1:6650
  components: {}
  config: {}

内部 Pulsar

inCluster 表示当 Milvus 集群启动时，集群中的 Pulsar 服务会自动启动。

示例

下面的示例配置了内部 Pulsar 服务。

apiVersion: milvus.io/v1alpha1
kind: Milvus
metadata:
  name: my-release
  labels:
    app: milvus
spec:
  dependencies:
    pulsar:
      inCluster:
        values:
          components:
            autorecovery: false
          zookeeper:
            replicaCount: 1
          bookkeeper:
            replicaCount: 1
            resoureces:
              limit:
                cpu: '4'
              memory: 8Gi
            requests:
              cpu: 200m
              memory: 512Mi
          broker:
            replicaCount: 1
            configData:
              ## Enable `autoSkipNonRecoverableData` since bookkeeper is running
              ## without persistence
              autoSkipNonRecoverableData: "true"
              managedLedgerDefaultEnsembleSize: "1"
              managedLedgerDefaultWriteQuorum: "1"
              managedLedgerDefaultAckQuorum: "1"
          proxy:
            replicaCount: 1
  components: {}
  config: {}

说明

该示例指定了 Pulsar 各组件的副本数量、Pulsar BookKeeper 的计算资源以及其他配置。

说明

在values.yaml 中查找配置内部 Pulsar 服务的完整配置项。如上例所示，根据需要在pulsar.inCluster.values 下添加配置项。

假设配置文件名为milvuscluster.yaml ，运行以下命令应用配置。

kubectl apply -f milvuscluster.yaml

配置 Kafka

Pulsar 是 Milvus 集群的默认消息存储。如果要使用 Kafka，请添加可选字段msgStreamType 以配置 Kafka。

kafka 支持external 和inCluster 。

外部 Kafka

external 表示使用外部 Kafka 服务。

用于配置外部 Kafka 服务的字段包括

external:true 值表示 Milvus 使用外部 Kafka 服务。
brokerList:要向其发送消息的代理列表。

示例

下面的示例配置了外部 Kafka 服务。

apiVersion: milvus.io/v1alpha1
kind: Milvus
metadata:
  name: my-release
  labels:
    app: milvus
spec:
  config:
    kafka:
      # securityProtocol supports: PLAINTEXT, SSL, SASL_PLAINTEXT, SASL_SSL
      securityProtocol: PLAINTEXT
      # saslMechanisms supports: PLAIN, SCRAM-SHA-256, SCRAM-SHA-512
      saslMechanisms: PLAIN
      saslUsername: ""
      saslPassword: ""
  # Omit other fields ...
  dependencies:
    # Omit other fields ...
    msgStreamType: "kafka"
    kafka:
      external: true
      brokerList:
        - "kafkaBrokerAddr1:9092"
        - "kafkaBrokerAddr2:9092"
        # ...

操作符 v0.8.5 或更高版本支持 SASL 配置。

内部 Kafka

inCluster 表示当 milvus 集群启动时，集群中的 Kafka 服务会自动启动。

示例

下面的示例配置了内部 Kafka 服务。

apiVersion: milvus.io/v1alpha1
kind: Milvus
metadata:
  name: my-release
  labels:
    app: milvus
spec:
  dependencies:
    msgStreamType: "kafka"
    kafka:
      inCluster:
        values: {} # values can be found in https://artifacthub.io/packages/helm/bitnami/kafka
  components: {}
  config: {}

在此处查找配置内部 Kafka 服务的完整配置项。根据需要在kafka.inCluster.values 下添加配置项。

假设配置文件名为milvuscluster.yaml ，运行以下命令应用配置。

kubectl apply -f milvuscluster.yaml

下一步

了解如何使用 Milvus Operator 配置其他 Milvus 依赖项：

开始之前​

配置 RocksMQ​

示例​

主要配置选项：​

配置 NATS​

示例​

配置 Pulsar​

外部 Pulsar​

示例​

内部 Pulsar​

示例​

配置 Kafka​

外部 Kafka​

示例​

内部 Kafka​

示例​

下一步​