Milvus

Milvus source connector

Description

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

• batch
• exactly-once
• column projection
• parallelism
• support multiple table read

Data Type Mapping

Milvus Data Type	SeaTunnel Data Type
INT8	TINYINT
INT16	SMALLINT
INT32	INT
INT64	BIGINT
FLOAT	FLOAT
DOUBLE	DOUBLE
BOOL	BOOLEAN
JSON	STRING
ARRAY	ARRAY
VARCHAR	STRING
FLOAT_VECTOR	FLOAT_VECTOR
BINARY_VECTOR	BINARY_VECTOR
FLOAT16_VECTOR	FLOAT16_VECTOR
BFLOAT16_VECTOR	BFLOAT16_VECTOR
SPARSE_FLOAT_VECTOR	SPARSE_FLOAT_VECTOR

Source Options

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 legacy alias collection_name is also accepted.
batch_size	Integer	No	1000	Number of records to fetch from Milvus in one batch. A larger value improves throughput but uses more memory; set it to a smaller value when records contain large vector payloads.
rate_limit	Integer	No	1000000	Maximum number of records the reader requests from Milvus per second. Use this to throttle a streaming job against the Milvus quota (QPS) or gRPC message-size limit. Set to -1 to disable throttling.

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.
• batch_size controls the per-fetch page size, not the parallelism of readers. Tune it together with parallelism to balance throughput and memory.
• rate_limit is a server-side hint that protects against Milvus GRPC limit errors when reading large volumes of vector data. Leave it at the default unless you see rate-limit or gRPC errors in the logs.
• When collection is not set, the source discovers all collections in database and exposes each collection as a separate SeaTunnel table.
• The source splits work by Milvus partition. Collections with a partition key are read with one split; collections without a partition key are split by partition name and assigned across readers.
• 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.
• Streaming jobs should set a checkpoint interval and reuse the same Milvus token across restarts so that incremental reads resume correctly from the committed offset.

Task Example

Read One Collection

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

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

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

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
  }
}

Stream From One Collection With Checkpoints

This example runs the source in STREAMING mode with a 30 second checkpoint interval. The downstream sink uses enable_upsert = false so each row is inserted once and duplicates are rejected.

env {
  parallelism = 1
  job.mode = "STREAMING"
  checkpoint.interval = 30000
}

source {
  Milvus {
    url = "http://127.0.0.1:19530"
    token = "username:password"
    database = "streaming_test"
    collection = "simple_example"
    batch_size = 500
    rate_limit = 200000
  }
}

sink {
  Milvus {
    url = "http://127.0.0.1:19530"
    token = "username:password"
    database = "streaming_test"
    enable_upsert = false
    batch_size = 1000
  }
}

Throttle Reads Against a Shared Cluster

When the Milvus cluster is shared with other jobs, lower rate_limit and batch_size so the source does not exceed the cluster's gRPC message-size limit.

env {
  parallelism = 2
  job.mode = "BATCH"
}

source {
  Milvus {
    url = "http://127.0.0.1:19530"
    token = "username:password"
    database = "shared"
    batch_size = 200
    rate_limit = 100000
  }
}

sink {
  Console {}
}

Changelog

Change Log

Change	Commit	Version
[Feature][Transform-V2] Support vector series sql function (#9765)	https://github.com/apache/seatunnel/commit/a40114cf7a	2.3.12
[Improve][Connector-milvus]update milvus-sdk-java to 2.5.11 (#9710)	https://github.com/apache/seatunnel/commit/08ebbaa8bd	2.3.12
[Chore] fix typos filed -> field (#9757)	https://github.com/apache/seatunnel/commit/e3e1c67d29	2.3.12
[Improve][Connector-V2] Optimize Milvus doc and e2e test case (#9766)	https://github.com/apache/seatunnel/commit/e67466f73e	2.3.12
[Improve][API] Optimize the enumerator API semantics and reduce lock calls at the connector level (#9671)	https://github.com/apache/seatunnel/commit/9212a77140	2.3.12
[Improve][API] Add metadata schema into catalog table (#9586)	https://github.com/apache/seatunnel/commit/385814e7f1	2.3.12
[Feature][Transform] Support define sink column type (#9114)	https://github.com/apache/seatunnel/commit/ab7119e507	2.3.11
[Feature][Checkpoint] Add check script for source/sink state class serialVersionUID missing (#9118)	https://github.com/apache/seatunnel/commit/4f5adeb1c7	2.3.11
[improve] milvus options (#9165)	https://github.com/apache/seatunnel/commit/5247e17640	2.3.11
[Fix][Connector-V2] Fix load state check in MilvusSourceReader to consider partition-level status (#8937)	https://github.com/apache/seatunnel/commit/bde235090b	2.3.10
[Improve][dist]add shade check rule (#8136)	https://github.com/apache/seatunnel/commit/51ef800016	2.3.9
[Improve][Core] Refactor common options of column/row (#7911)	https://github.com/apache/seatunnel/commit/d1582afee6	2.3.9
[Feature][connector-milvus] update milvus connector to support dynamic schema, failed retry, etc. (#7885)	https://github.com/apache/seatunnel/commit/6a31f91729	2.3.9
[Feature][Restapi] Allow metrics information to be associated to logical plan nodes (#7786)	https://github.com/apache/seatunnel/commit/6b7c53d03c	2.3.9
[Fix][Connector-V2] Fix known directory create and delete ignore issues (#7700)	https://github.com/apache/seatunnel/commit/e2fb679577	2.3.8
[Improve][Connector-V2] Optimize milvus code (#7691)	https://github.com/apache/seatunnel/commit/1eddb8e1b1	2.3.8
[Improve][Connector-V2] Optimize milvus-connector config code (#7658)	https://github.com/apache/seatunnel/commit/f831f7a5ec	2.3.8
[Improve][Connector-V2] update vectorType (#7446)	https://github.com/apache/seatunnel/commit/1bba72385b	2.3.8
[Improve][API] Move catalog open to SaveModeHandler (#7439)	https://github.com/apache/seatunnel/commit/8c2c5c79a1	2.3.8
[Feature][Connector-V2] Fake Source support produce vector data (#7401)	https://github.com/apache/seatunnel/commit/6937d10ac3	2.3.8
[Feature][Connector-V2][Milvus] Support Milvus source & sink (#7158)	https://github.com/apache/seatunnel/commit/0c69b9166e	2.3.6