Milvus

Milvus source connector

Support Those Engines

Spark
Flink
SeaTunnel Zeta

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	Server-side query rate limit (QPS) applied to the source collection via the Milvus collection.queryRate.max.qps property. The reader mutates this collection-wide setting while the job is running, so it affects every client of the collection, not just this SeaTunnel job. Set to -1 to disable.

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 mutates the server-side collection.queryRate.max.qps property on every collection the job reads, so the new limit applies to all clients of that collection while the job is running. The reader resets the property to -1 on close, but a job crash before close() will leave the collection throttled until it is restored manually. Leave it at the default unless you observe throttling 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.
• The Milvus source is BOUNDED: the job finishes naturally once every partition (split) has been fully scanned, and unlike Kafka or Fluss there is no per-record offset for continuous incremental reads. Checkpoint/restore is at split (partition) granularity — a partition that has already been fully scanned is not re-read, but a partition that was in progress when the job failed is re-scanned from the beginning of that partition. If you need to keep ingesting newly written vectors, re-submit the SeaTunnel job periodically from an external scheduler.

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

Re-submit with Checkpoints for Periodic Ingestion

The Milvus source is BOUNDED, so the job finishes naturally once every partition has been fully scanned. This example runs in STREAMING mode with a short checkpoint interval — to keep ingesting newly written vectors, re-submit the job from an external scheduler on demand. On restore, recovery is at split (partition) granularity: partitions that were already fully scanned are not re-read, while partitions that were in progress when the job failed are re-scanned from the beginning of that partition. The downstream sink uses enable_upsert = true and dedupes by primary key to avoid duplicate writes on re-scanned partitions.

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 = true
    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 per-collection query quota.

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

FAQ

Can Milvus source read all collections in a database at once?

Yes. If you omit the collection parameter or leave it empty, the Milvus source connector will scan and read all collections in the configured database.

Which vector data types are supported?

The connector supports FLOAT_VECTOR, BINARY_VECTOR, FLOAT16_VECTOR, BFLOAT16_VECTOR, and SPARSE_FLOAT_VECTOR types, carrying index and partition metadata to downstream connectors.

How does the source handle gRPC message size or rate limits?

You can tune batch_size and rate_limit options. If the Milvus cluster enforces rate limits or gRPC message limits, the connector automatically retries with backoff.

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