This is an automated email from the ASF dual-hosted git repository.
zhangshenghang pushed a commit to branch dev
in repository https://gitbox.apache.org/repos/asf/seatunnel.git
The following commit(s) were added to refs/heads/dev by this push:
new 41fba790fd [Docs][Connector-V2] Improve Milvus connector docs (#11254)
41fba790fd is described below
commit 41fba790fda61b94cf93f84f3a8f6e4062958672
Author: Daniel Carter <[email protected]>
AuthorDate: Fri Jul 3 08:12:56 2026 +0800
[Docs][Connector-V2] Improve Milvus connector docs (#11254)
Co-authored-by: DanielCarter-stack
<[email protected]>
Co-authored-by: Shenghang <[email protected]>
---
docs/en/connectors/sink/Milvus.md | 246 +++++++++++++++++++++++++++++----
docs/en/connectors/source/Milvus.md | 124 +++++++++++++++--
docs/zh/connectors/sink/Milvus.md | 262 +++++++++++++++++++++++++++++++-----
docs/zh/connectors/source/Milvus.md | 127 +++++++++++++++--
4 files changed, 670 insertions(+), 89 deletions(-)
diff --git a/docs/en/connectors/sink/Milvus.md
b/docs/en/connectors/sink/Milvus.md
index 434fec2084..86d9bae581 100644
--- a/docs/en/connectors/sink/Milvus.md
+++ b/docs/en/connectors/sink/Milvus.md
@@ -6,11 +6,18 @@ import ChangeLog from '../changelog/connector-milvus.md';
## Description
-This Milvus sink connector write data to Milvus or Zilliz Cloud, it has the
following features:
-- support read and write data by partition
-- support write dynamic schema data from Metadata Column
-- json data will be converted to json string and sink as json as well
-- retry automatically to bypass ratelimit and grpc limit
+This Milvus sink connector writes data to Milvus or Zilliz Cloud. It can
create the target
+collection from the upstream SeaTunnel schema and supports vector fields,
dynamic fields,
+partition keys, partition metadata, and automatic retries for rate limit or
gRPC limit errors.
+
+Common use cases:
+
+- Write `FLOAT_VECTOR`, `BINARY_VECTOR`, `FLOAT16_VECTOR`, `BFLOAT16_VECTOR`,
and `SPARSE_FLOAT_VECTOR` fields.
+- Create a collection when it does not exist.
+- Copy data from one Milvus collection to another Milvus collection.
+- Preserve partition information when the upstream Milvus source carries
partition metadata.
+- Create vector indexes and load the collection after writing when requested.
+
## Key Features
- [x] [batch](../../introduction/concepts/connector-v2-features.md)
@@ -40,51 +47,234 @@ This Milvus sink connector write data to Milvus or Zilliz
Cloud, it has the foll
## Sink Options
-| Name | Type | Required | Default
| Description
|
-|------------------------|---------------------|----------|------------------------------|-----------------------------------------------------------------------------------------------------------------------------------------------------|
-| url | String | Yes | -
| The URL to connect to Milvus or Zilliz Cloud.
|
-| token | String | Yes | -
| User:password
|
-| database | String | No | -
| Write data to which database, default is source database.
|
-| collection | String | No | -
| Write data to which collection, default is source table name. The
deprecated `collection_name` key is accepted as an alias.
|
-| schema_save_mode | enum | No |
CREATE_SCHEMA_WHEN_NOT_EXIST | Auto create table when table not exist.
|
-| enable_auto_id | boolean | No | false
| Primary key column enable autoId.
|
-| enable_upsert | boolean | No | false
| Upsert data not insert.
|
-| enable_dynamic_field | boolean | No | true
| Enable create table with dynamic field.
|
-| batch_size | int | No | 1000
| Write batch size. When the number of buffered records reaches
`batch_size` or the time reaches `checkpoint.interval`, it will trigger a write
flush |
-| partition_key | String | No |
| Milvus partition key field
|
-| create_index | boolean | No | false
| Automatically create vector indexes for collection to improve query
performance.
|
-| load_collection | boolean | No | false
| Load collection into Milvus memory for immediate query
availability.
|
-| collection_description | Map<String, String> | No | {}
| Collection descriptions map where key is collection name and value
is description.
|
+| Name | Type | Required | Default
| Description
|
+|------------------------|---------------------|----------|------------------------------|----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|
+| url | String | Yes | -
| The URL to connect to Milvus or Zilliz Cloud, for example
`http://127.0.0.1:19530`.
|
+| token | String | Yes | -
| Milvus authentication token. For a local Milvus server this is
usually `username:password`.
|
+| database | String | No | -
| Target database. If it is not set, the sink uses the upstream
database when available.
|
+| collection | String | No | -
| Target collection. If it is not set, the sink uses the upstream
table name. The deprecated `collection_name` key is accepted as an alias.
|
+| schema_save_mode | enum | No |
CREATE_SCHEMA_WHEN_NOT_EXIST | Controls how SeaTunnel handles the target
collection schema. The default creates the collection only when it does not
already exist.
|
+| data_save_mode | enum | No | APPEND_DATA
| Controls how SeaTunnel handles existing data. Supported values are
`DROP_DATA`, `APPEND_DATA`, and `ERROR_WHEN_DATA_EXISTS`.
|
+| enable_auto_id | boolean | No | false
| Whether Milvus should generate primary key values automatically. If
this is `true`, do not send values for the primary key field.
|
+| enable_upsert | boolean | No | true
| Whether to write by upsert instead of insert. Upsert needs a
primary key in the collection schema.
|
+| enable_dynamic_field | boolean | No | true
| Whether to enable Milvus dynamic fields when SeaTunnel creates a
collection.
|
+| batch_size | int | No | 1000
| Number of records buffered before one write. In streaming jobs,
data can also be flushed when a checkpoint is triggered.
|
+| rate_limit | int | No | 100000
| Maximum write rate used by the connector retry/rate-limit control.
|
+| partition_key | String | No | -
| Field name used as the Milvus partition key when SeaTunnel creates
a collection.
|
+| create_index | boolean | No | false
| Whether to create vector indexes for the target collection. When
copying from Milvus source, existing vector index metadata can be used to
create matching indexes on the target collection.
|
+| load_collection | boolean | No | false
| Whether to load the target collection into Milvus memory after the
collection is created. This is useful when the data should be queried
immediately after writing.
|
+| collection_description | Map<String, String> | No | {}
| Collection descriptions map. The key is the collection name and the
value is the description used when the collection is created.
|
+
+## Notes
+
+- Vector dimensions come from the SeaTunnel schema. For vector fields
generated by `FakeSource`, `columnScale` is used as the vector dimension.
+- If `collection` is not set, the sink uses the upstream table name as the
Milvus collection name. This is useful for multi-table jobs.
+- When a Milvus source reads partitions, the sink can create the same
partition names on the target collection if the target collection does not use
a partition key.
+- `create_index = true` only creates vector indexes when index metadata is
available from the upstream catalog, such as when the source is Milvus.
## Task Example
-### Basic Configuration
+### Write a SeaTunnel Schema to Milvus
+
+This example writes 10 rows into the `test1.simple_example_1` collection. The
sink uses the source
+table name as the collection name because `collection` is not set.
+
```bash
+env {
+ parallelism = 1
+ job.mode = "BATCH"
+}
+
+source {
+ FakeSource {
+ row.num = 10
+ vector.dimension = 4
+ schema = {
+ table = "simple_example_1"
+ columns = [
+ {
+ name = book_id
+ type = bigint
+ nullable = false
+ defaultValue = 0
+ comment = "primary key id"
+ },
+ {
+ name = book_intro
+ type = float_vector
+ columnScale = 4
+ comment = "vector"
+ },
+ {
+ name = book_title
+ type = string
+ nullable = true
+ comment = "topic"
+ }
+ ]
+ primaryKey {
+ name = book_id
+ columnNames = [book_id]
+ }
+ }
+ }
+}
+
sink {
Milvus {
url = "http://127.0.0.1:19530";
token = "username:password"
- collection = "user_vectors"
- batch_size = 1000
+ database = "test1"
}
}
```
-### Advanced Configuration with Index and Loading
+### Write Multiple Vector Types
+
```bash
+env {
+ parallelism = 1
+ job.mode = "BATCH"
+}
+
+source {
+ FakeSource {
+ row.num = 10
+ vector.dimension = 4
+ binary.vector.dimension = 8
+ schema = {
+ table = "simple_example_2"
+ columns = [
+ {
+ name = book_id
+ type = bigint
+ nullable = false
+ defaultValue = 0
+ comment = "primary key id"
+ },
+ {
+ name = book_intro_1
+ type = binary_vector
+ columnScale = 8
+ comment = "binary vector"
+ },
+ {
+ name = book_intro_2
+ type = float16_vector
+ columnScale = 4
+ comment = "float16 vector"
+ },
+ {
+ name = book_intro_3
+ type = bfloat16_vector
+ columnScale = 4
+ comment = "bfloat16 vector"
+ },
+ {
+ name = book_intro_4
+ type = sparse_float_vector
+ columnScale = 4
+ comment = "sparse vector"
+ }
+ ]
+ primaryKey {
+ name = book_id
+ columnNames = [book_id]
+ }
+ }
+ }
+}
+
sink {
Milvus {
url = "http://127.0.0.1:19530";
token = "username:password"
- batch_size = 1000
+ database = "test2"
+ }
+}
+```
+
+### Copy One Milvus Collection and Create Indexes
+
+```bash
+env {
+ parallelism = 1
+ job.mode = "BATCH"
+}
+
+source {
+ Milvus {
+ url = "http://127.0.0.1:19530";
+ token = "username:password"
+ collection = "simple_example"
+ }
+}
+
+sink {
+ Milvus {
+ url = "http://127.0.0.1:19530";
+ token = "username:password"
+ database = "test_index_preservation"
+ collection = "simple_example_preservation"
create_index = true
load_collection = true
- collection_description = {
- "user_vectors" = "User embedding vectors for recommendation"
- "product_vectors" = "Product feature vectors for search"
+ }
+}
+```
+
+### Streaming Write with Checkpoint Flush
+
+```bash
+env {
+ parallelism = 1
+ job.mode = "STREAMING"
+ checkpoint.interval = 30000
+}
+
+source {
+ FakeSource {
+ row.num = 10
+ vector.dimension = 4
+ schema = {
+ table = "streaming_simple_example"
+ columns = [
+ {
+ name = book_id
+ type = bigint
+ nullable = false
+ defaultValue = 0
+ comment = "primary key id"
+ },
+ {
+ name = book_intro
+ type = float_vector
+ columnScale = 4
+ comment = "vector"
+ },
+ {
+ name = book_title
+ type = string
+ nullable = true
+ comment = "topic"
+ }
+ ]
+ primaryKey {
+ name = book_id
+ columnNames = [book_id]
+ }
}
}
}
+
+sink {
+ Milvus {
+ url = "http://127.0.0.1:19530";
+ token = "username:password"
+ database = "streaming_test"
+ enable_upsert = false
+ batch_size = 3
+ }
+}
```
## Changelog
diff --git a/docs/en/connectors/source/Milvus.md
b/docs/en/connectors/source/Milvus.md
index 9ac2cd947d..d90bbcc773 100644
--- a/docs/en/connectors/source/Milvus.md
+++ b/docs/en/connectors/source/Milvus.md
@@ -6,11 +6,17 @@ import ChangeLog from '../changelog/connector-milvus.md';
## Description
-This Milvus source connector reads data from Milvus or Zilliz Cloud, it has
the following features:
-- support read and write data by partition
-- support read dynamic schema data into Metadata Column
-- json data will be converted to json string and sink as json as well
-- retry automatically to bypass ratelimit and grpc limit
+This Milvus source connector reads data from Milvus or Zilliz Cloud. It can
read one collection
+or all collections in a database, and it carries Milvus metadata such as
partition information and
+vector index information to downstream connectors when the target connector
can use it.
+
+Common use cases:
+
+- Read one Milvus collection by setting `collection`.
+- Read all collections in a Milvus database by leaving `collection` empty.
+- Copy data from Milvus to Milvus while preserving vector fields, partition
metadata, and index metadata.
+- Read `FLOAT_VECTOR`, `BINARY_VECTOR`, `FLOAT16_VECTOR`, `BFLOAT16_VECTOR`,
and `SPARSE_FLOAT_VECTOR` fields.
+- Retry automatically to bypass rate limit or gRPC limit errors.
## Key Features
@@ -40,21 +46,116 @@ This Milvus source connector reads data from Milvus or
Zilliz Cloud, it has the
## Source Options
-| Name | Type | Required | Default |
Description |
-|------------|--------|----------|---------|--------------------------------------------------------------------------------------------|
-| url | String | Yes | - | The URL to connect to Milvus or
Zilliz Cloud. |
-| token | String | Yes | - | User:password
|
-| database | String | Yes | default | Read data from which database.
|
-| collection | String | No | - | If set, will only read one
collection, otherwise will read all collections under database. |
+| Name | Type | Required | Default | Description
|
+|------------|--------|----------|---------|----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|
+| url | String | Yes | - | The URL to connect to Milvus or
Zilliz Cloud, for example `http://127.0.0.1:19530`.
|
+| token | String | Yes | - | Milvus authentication token. For
a local Milvus server this is usually `username:password`.
|
+| database | String | No | default | Source database.
|
+| collection | String | No | - | Source collection. If it is set,
only this collection is read. If it is not set, all collections under
`database` are read. The deprecated `collection_name` key is accepted as an
alias. |
+| batch_size | Int | No | 1000 | Number of records fetched in one
source read batch.
|
+| rate_limit | Int | No | 1000000 | Maximum source read rate limit.
|
+
+## Notes
+
+- `database` defaults to `default`, so simple local Milvus jobs do not need to
set it.
+- `collection` is optional. Set it when the job should read exactly one
collection.
+- When the source reads a collection with partitions, downstream Milvus sink
can use that metadata to create the same partition names on the target
collection.
+- When the source reads vector indexes, downstream Milvus sink can use that
metadata with `create_index = true` to create matching vector indexes.
## Task Example
+### Read One Collection
+
```bash
+env {
+ parallelism = 1
+ job.mode = "BATCH"
+}
+
source {
Milvus {
url = "http://127.0.0.1:19530";
token = "username:password"
database = "default"
+ collection = "simple_example"
+ }
+}
+
+sink {
+ Console {}
+}
+```
+
+### Read All Collections in a Database
+
+```bash
+env {
+ parallelism = 1
+ job.mode = "BATCH"
+}
+
+source {
+ Milvus {
+ url = "http://127.0.0.1:19530";
+ token = "username:password"
+ database = "default"
+ }
+}
+
+sink {
+ Console {}
+}
+```
+
+### Copy a Milvus Collection to Another Database
+
+```bash
+env {
+ parallelism = 1
+ job.mode = "BATCH"
+}
+
+source {
+ Milvus {
+ url = "http://127.0.0.1:19530";
+ token = "username:password"
+ collection = "simple_example"
+ }
+}
+
+sink {
+ Milvus {
+ url = "http://127.0.0.1:19530";
+ token = "username:password"
+ database = "test"
+ collection = "simple_example"
+ }
+}
+```
+
+### Copy a Collection and Recreate Vector Indexes
+
+```bash
+env {
+ parallelism = 1
+ job.mode = "BATCH"
+}
+
+source {
+ Milvus {
+ url = "http://127.0.0.1:19530";
+ token = "username:password"
+ collection = "simple_example"
+ }
+}
+
+sink {
+ Milvus {
+ url = "http://127.0.0.1:19530";
+ token = "username:password"
+ database = "test_index_preservation"
+ collection = "simple_example_preservation"
+ create_index = true
}
}
```
@@ -62,4 +163,3 @@ source {
## Changelog
<ChangeLog />
-
diff --git a/docs/zh/connectors/sink/Milvus.md
b/docs/zh/connectors/sink/Milvus.md
index 33bb0fd75e..7738d2918d 100644
--- a/docs/zh/connectors/sink/Milvus.md
+++ b/docs/zh/connectors/sink/Milvus.md
@@ -6,18 +6,26 @@ import ChangeLog from '../changelog/connector-milvus.md';
## 描述
-Milvus sink连接器将数据写入Milvus或Zilliz Cloud，它具有以下功能：
-- 支持按分区读写数据
-- 支持从元数据列写入动态模式数据
-- json数据将转换为json字符串进行写入
-- 自动重试以绕过 ratelimit 限制 和 grpc 限制
+Milvus 接收器用于把数据写入 Milvus 或 Zilliz Cloud。它可以根据上游
+SeaTunnel 表结构创建目标集合，并支持向量字段、动态字段、分区键、分区元数据，以及
+遇到限流或 gRPC 限制时自动重试。
+
+常见用法：
+
+- 写入 `FLOAT_VECTOR`、`BINARY_VECTOR`、`FLOAT16_VECTOR`、`BFLOAT16_VECTOR` 和
`SPARSE_FLOAT_VECTOR` 字段。
+- 目标集合不存在时自动创建集合。
+- 从一个 Milvus 集合复制数据到另一个 Milvus 集合。
+- 上游 Milvus 源连接器携带分区信息时，在目标集合保留分区。
+- 按需创建向量索引，并在写入后加载集合。
+
## 主要特性
-- [x] [batch](../../introduction/concepts/connector-v2-features.md)
-- [x] [exactly-once](../../introduction/concepts/connector-v2-features.md)
-- [ ] [column projection](../../introduction/concepts/connector-v2-features.md)
+- [x] [批处理](../../introduction/concepts/connector-v2-features.md)
+- [x] [精确一次](../../introduction/concepts/connector-v2-features.md)
+- [ ] [列投影](../../introduction/concepts/connector-v2-features.md)
+- [ ] [定时刷新](../../introduction/concepts/connector-v2-features.md)
-##数据类型映射
+## 数据类型映射
| Milvus数据类型 | SeaTunnel 数据类型 |
|---------------------|---------------------|
@@ -36,55 +44,237 @@ Milvus sink连接器将数据写入Milvus或Zilliz Cloud，它具有以下功能
| FLOAT16_VECTOR | FLOAT16_VECTOR |
| BFLOAT16_VECTOR | BFLOAT16_VECTOR |
| SPARSE_FLOAT_VECTOR | SPARSE_FLOAT_VECTOR |
-- [ ] [timer flush](../../introduction/concepts/connector-v2-features.md)
-
-## Sink 选项
-
-| 名字 | 类型 | 是否必传 | 默认值
| 描述 |
-|------------------------|---------------------|------|------------------------------|---------------------------------------------------------------------|
-| url | String | 是 | -
| 连接到Milvus或Zilliz Cloud的URL。 |
-| token | String | 是 | -
| 用户：密码 |
-| database | String | 否 | -
| 将数据写入哪个数据库，默认为源数据库。 |
-| collection | String | 否 | -
| 将数据写入哪个集合，默认为源表名。兼容旧配置键 `collection_name`。
|
-| schema_save_mode | enum | 否 |
CREATE_SCHEMA_WHEN_NOT_EXIST | 当表不存在时自动创建表。
|
-| enable_auto_id | boolean | 否 | false
| 主键列启用autoId。 |
-| enable_upsert | boolean | 否 | false
| 是否启用upsert。 |
-| enable_dynamic_field | boolean | 否 | true
| 是否启用带动态字段的创建表。 |
-| batch_size | int | 否 | 1000
| 写入批大小。当缓冲记录数达到 `batch_size` 或时间达到 `checkpoint.interval` 时，将触发一次写入刷新 |
-| partition_key | String | 否 |
| Milvus分区键字段 |
-| create_index | boolean | No | false
| 自动为集合创建向量索引以提高查询性能 |
-| load_collection | boolean | No | false
| 将集合加载到 Milvus 内存中以便立即进行查询 |
-| collection_description | Map<String, String> | No | {}
| 集合描述映射，其中键是集合名称，值是描述 |
+
+## 接收器选项
+
+| 名字 | 类型 | 是否必传 | 默认值
| 描述
|
+|------------------------|---------------------|------|------------------------------|-----------------------------------------------------------------------------------------|
+| url | String | 是 | -
| Milvus 或 Zilliz Cloud 的连接地址，例如 `http://127.0.0.1:19530`。
|
+| token | String | 是 | -
| Milvus 认证令牌。本地 Milvus 通常使用 `username:password`。
|
+| database | String | 否 | -
| 目标数据库。不填时，如果上游带有数据库信息，则使用上游数据库。
|
+| collection | String | 否 | -
| 目标集合。不填时使用上游表名作为集合名。兼容旧配置键 `collection_name`。
|
+| schema_save_mode | enum | 否 |
CREATE_SCHEMA_WHEN_NOT_EXIST | 控制如何处理目标集合结构。默认值表示集合不存在时才创建集合。
|
+| data_save_mode | enum | 否 | APPEND_DATA
| 控制如何处理已有数据。支持 `DROP_DATA`、`APPEND_DATA`、`ERROR_WHEN_DATA_EXISTS`。
|
+| enable_auto_id | boolean | 否 | false
| 是否让 Milvus 自动生成主键。设置为 `true` 时，不要再向主键字段写值。
|
+| enable_upsert | boolean | 否 | true
| 是否使用更新插入模式写入，而不是普通插入模式。更新插入需要集合中有主键。
|
+| enable_dynamic_field | boolean | 否 | true
| SeaTunnel 创建集合时，是否启用 Milvus 动态字段。
|
+| batch_size | int | 否 | 1000
| 写入前缓存的记录数。流任务中，检查点触发时也会刷新写入。
|
+| rate_limit | int | 否 | 100000
| 连接器用于重试和限流控制的最大写入速率。
|
+| partition_key | String | 否 | -
| SeaTunnel 创建集合时使用的 Milvus 分区键字段名。
|
+| create_index | boolean | 否 | false
| 是否为目标集合创建向量索引。从 Milvus 源端复制时，可以利用上游索引元数据在目标集合创建相同索引。
|
+| load_collection | boolean | 否 | false
| 创建集合后是否把目标集合加载到 Milvus 内存中。如果写完马上要查询，可以开启。
|
+| collection_description | Map<String, String> | 否 | {}
| 集合描述映射。键是集合名，值是创建集合时写入的描述。
|
+
+## 注意事项
+
+- 向量维度来自 SeaTunnel 表结构。使用 `FakeSource` 生成向量字段时，`columnScale` 表示向量维度。
+- 不配置 `collection` 时，接收器会使用上游表名作为 Milvus 集合名，适合多表写入。
+- Milvus 源端读取分区后，如果目标集合没有使用分区键，接收器可以在目标集合创建相同分区名。
+- `create_index = true` 只有在上游目录信息能提供索引元数据时才会创建向量索引，例如上游也是 Milvus。
## 任务示例
-### 基础配置
+### 写入 SeaTunnel 表结构到 Milvus
+
+这个示例会把 10 行数据写入 `test1.simple_example_1` 集合。因为没有配置
+`collection`，接收器会使用源端的表名作为集合名。
+
```bash
+env {
+ parallelism = 1
+ job.mode = "BATCH"
+}
+
+source {
+ FakeSource {
+ row.num = 10
+ vector.dimension = 4
+ schema = {
+ table = "simple_example_1"
+ columns = [
+ {
+ name = book_id
+ type = bigint
+ nullable = false
+ defaultValue = 0
+ comment = "primary key id"
+ },
+ {
+ name = book_intro
+ type = float_vector
+ columnScale = 4
+ comment = "vector"
+ },
+ {
+ name = book_title
+ type = string
+ nullable = true
+ comment = "topic"
+ }
+ ]
+ primaryKey {
+ name = book_id
+ columnNames = [book_id]
+ }
+ }
+ }
+}
+
sink {
Milvus {
url = "http://127.0.0.1:19530";
token = "username:password"
- collection = "user_vectors"
- batch_size = 1000
+ database = "test1"
}
}
```
-### 带 Index 和 Loading 的高级配置
+### 写入多种向量类型
+
```bash
+env {
+ parallelism = 1
+ job.mode = "BATCH"
+}
+
+source {
+ FakeSource {
+ row.num = 10
+ vector.dimension = 4
+ binary.vector.dimension = 8
+ schema = {
+ table = "simple_example_2"
+ columns = [
+ {
+ name = book_id
+ type = bigint
+ nullable = false
+ defaultValue = 0
+ comment = "primary key id"
+ },
+ {
+ name = book_intro_1
+ type = binary_vector
+ columnScale = 8
+ comment = "binary vector"
+ },
+ {
+ name = book_intro_2
+ type = float16_vector
+ columnScale = 4
+ comment = "float16 vector"
+ },
+ {
+ name = book_intro_3
+ type = bfloat16_vector
+ columnScale = 4
+ comment = "bfloat16 vector"
+ },
+ {
+ name = book_intro_4
+ type = sparse_float_vector
+ columnScale = 4
+ comment = "sparse vector"
+ }
+ ]
+ primaryKey {
+ name = book_id
+ columnNames = [book_id]
+ }
+ }
+ }
+}
+
sink {
Milvus {
url = "http://127.0.0.1:19530";
token = "username:password"
- batch_size = 1000
+ database = "test2"
+ }
+}
+```
+
+### 复制一个 Milvus 集合并创建索引
+
+```bash
+env {
+ parallelism = 1
+ job.mode = "BATCH"
+}
+
+source {
+ Milvus {
+ url = "http://127.0.0.1:19530";
+ token = "username:password"
+ collection = "simple_example"
+ }
+}
+
+sink {
+ Milvus {
+ url = "http://127.0.0.1:19530";
+ token = "username:password"
+ database = "test_index_preservation"
+ collection = "simple_example_preservation"
create_index = true
load_collection = true
- collection_description = {
- "user_vectors" = "User embedding vectors for recommendation"
- "product_vectors" = "Product feature vectors for search"
+ }
+}
+```
+
+### 流式写入并通过检查点刷新
+
+```bash
+env {
+ parallelism = 1
+ job.mode = "STREAMING"
+ checkpoint.interval = 30000
+}
+
+source {
+ FakeSource {
+ row.num = 10
+ vector.dimension = 4
+ schema = {
+ table = "streaming_simple_example"
+ columns = [
+ {
+ name = book_id
+ type = bigint
+ nullable = false
+ defaultValue = 0
+ comment = "primary key id"
+ },
+ {
+ name = book_intro
+ type = float_vector
+ columnScale = 4
+ comment = "vector"
+ },
+ {
+ name = book_title
+ type = string
+ nullable = true
+ comment = "topic"
+ }
+ ]
+ primaryKey {
+ name = book_id
+ columnNames = [book_id]
+ }
}
}
}
+
+sink {
+ Milvus {
+ url = "http://127.0.0.1:19530";
+ token = "username:password"
+ database = "streaming_test"
+ enable_upsert = false
+ batch_size = 3
+ }
+}
```
## 变更日志
diff --git a/docs/zh/connectors/source/Milvus.md
b/docs/zh/connectors/source/Milvus.md
index bdf36bfedf..2c6e2bf39c 100644
--- a/docs/zh/connectors/source/Milvus.md
+++ b/docs/zh/connectors/source/Milvus.md
@@ -6,13 +6,19 @@ import ChangeLog from '../changelog/connector-milvus.md';
## 描述
-这个Milvus源连接器从Milvus或Zilliz Cloud读取数据，它具有以下功能：
-- 支持按分区读写数据
-- 支持将动态模式数据读入元数据列
-- json数据将转换为json字符串，并将sink转换为json
-- 自动重试以绕过速率限制和grpc限制
+Milvus 源连接器用于从 Milvus 或 Zilliz Cloud 读取数据。它可以读取一个集合，
+也可以读取某个数据库下的所有集合，并且会把 Milvus 的分区信息、向量索引信息等元数据传给下游，
+下游连接器支持时可以继续使用这些信息。
-## 关键特性
+常见用法：
+
+- 配置 `collection` 读取一个 Milvus 集合。
+- 不配置 `collection` 时，读取指定数据库下的所有集合。
+- 从 Milvus 复制到 Milvus，并保留向量字段、分区元数据和索引元数据。
+- 读取 `FLOAT_VECTOR`、`BINARY_VECTOR`、`FLOAT16_VECTOR`、`BFLOAT16_VECTOR` 和
`SPARSE_FLOAT_VECTOR` 字段。
+- 遇到限流或 gRPC 限制时自动重试。
+
+## 主要特性
- [x] [批处理](../../introduction/concepts/connector-v2-features.md)
- [x] [精确一次](../../introduction/concepts/connector-v2-features.md)
@@ -40,16 +46,54 @@ import ChangeLog from '../changelog/connector-milvus.md';
## 源选项
-| 名称 | 类型 | 必需 | 默认值 |
描述 |
-|------------|--------|----------|---------|--------------------------------------------------------------------------------------------|
-| url | String | 是 | - | 连接到Milvus或Zilliz Cloud的URL.
|
-| token | String | 是 | - | 用户：密码
|
-| database | String | 是 | default | 从哪个数据库读取数据.
|
-| collection | String | 否 | - | 如果设置，将只读取一个集合，否则将读取数据库下的所有集合. |
+| 名称 | 类型 | 是否必传 | 默认值 | 描述
|
+|------------|--------|------|---------|--------------------------------------------------------------------------------------------|
+| url | String | 是 | - | Milvus 或 Zilliz Cloud 的连接地址，例如
`http://127.0.0.1:19530`。 |
+| token | String | 是 | - | Milvus 认证令牌。本地 Milvus 通常使用
`username:password`。 |
+| database | String | 否 | default | 源数据库。
|
+| collection | String | 否 | - | 源集合。配置后只读取这个集合；不配置时读取 `database`
下的所有集合。兼容旧配置键 `collection_name`。 |
+| batch_size | Int | 否 | 1000 | 每批从 Milvus 读取的数据条数。
|
+| rate_limit | Int | 否 | 1000000 | 源端读取速率限制。
|
+
+## 注意事项
+
+- `database` 默认是 `default`，本地 Milvus 的简单任务通常不用配置。
+- `collection` 是可选项。只想读一个集合时再配置。
+- 源端读取带分区的集合时，下游 Milvus 接收器可以利用这些元数据在目标集合创建相同分区名。
+- 源端读取到向量索引信息时，下游 Milvus 接收器可以配合 `create_index = true` 创建相同向量索引。
## 任务示例
+### 读取一个集合
+
+```bash
+env {
+ parallelism = 1
+ job.mode = "BATCH"
+}
+
+source {
+ Milvus {
+ url = "http://127.0.0.1:19530";
+ token = "username:password"
+ database = "default"
+ collection = "simple_example"
+ }
+}
+
+sink {
+ Console {}
+}
+```
+
+### 读取一个数据库下的所有集合
+
```bash
+env {
+ parallelism = 1
+ job.mode = "BATCH"
+}
+
source {
Milvus {
url = "http://127.0.0.1:19530";
@@ -57,8 +101,65 @@ source {
database = "default"
}
}
+
+sink {
+ Console {}
+}
+```
+
+### 复制一个 Milvus 集合到另一个数据库
+
+```bash
+env {
+ parallelism = 1
+ job.mode = "BATCH"
+}
+
+source {
+ Milvus {
+ url = "http://127.0.0.1:19530";
+ token = "username:password"
+ collection = "simple_example"
+ }
+}
+
+sink {
+ Milvus {
+ url = "http://127.0.0.1:19530";
+ token = "username:password"
+ database = "test"
+ collection = "simple_example"
+ }
+}
+```
+
+### 复制集合并重建向量索引
+
+```bash
+env {
+ parallelism = 1
+ job.mode = "BATCH"
+}
+
+source {
+ Milvus {
+ url = "http://127.0.0.1:19530";
+ token = "username:password"
+ collection = "simple_example"
+ }
+}
+
+sink {
+ Milvus {
+ url = "http://127.0.0.1:19530";
+ token = "username:password"
+ database = "test_index_preservation"
+ collection = "simple_example_preservation"
+ create_index = true
+ }
+}
```
## 变更日志
-<ChangeLog />
\ No newline at end of file
+<ChangeLog />
