Skip to main content
Version: Next

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​

Data Type Mapping​

Milvus Data TypeSeaTunnel Data Type
INT8TINYINT
INT16SMALLINT
INT32INT
INT64BIGINT
FLOATFLOAT
DOUBLEDOUBLE
BOOLBOOLEAN
JSONSTRING
ARRAYARRAY
VARCHARSTRING
FLOAT_VECTORFLOAT_VECTOR
BINARY_VECTORBINARY_VECTOR
FLOAT16_VECTORFLOAT16_VECTOR
BFLOAT16_VECTORBFLOAT16_VECTOR
SPARSE_FLOAT_VECTORSPARSE_FLOAT_VECTOR

Source Options​

NameTypeRequiredDefaultDescription
urlStringYes-The URL to connect to Milvus or Zilliz Cloud, for example http://127.0.0.1:19530.
tokenStringYes-Milvus authentication token. For a local Milvus server this is usually username:password.
databaseStringNodefaultSource database.
collectionStringNo-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_sizeIntegerNo1000Number 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_limitIntegerNo1000000Server-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 {}
}

Changelog​

Change Log
ChangeCommitVersion
[Feature][Transform-V2] Support vector series sql function (#9765)https://github.com/apache/seatunnel/commit/a40114cf7a2.3.12
[Improve][Connector-milvus]update milvus-sdk-java to 2.5.11 (#9710)https://github.com/apache/seatunnel/commit/08ebbaa8bd2.3.12
[Chore] fix typos filed -> field (#9757)https://github.com/apache/seatunnel/commit/e3e1c67d292.3.12
[Improve][Connector-V2] Optimize Milvus doc and e2e test case (#9766)https://github.com/apache/seatunnel/commit/e67466f73e2.3.12
[Improve][API] Optimize the enumerator API semantics and reduce lock calls at the connector level (#9671)https://github.com/apache/seatunnel/commit/9212a771402.3.12
[Improve][API] Add metadata schema into catalog table (#9586)https://github.com/apache/seatunnel/commit/385814e7f12.3.12
[Feature][Transform] Support define sink column type (#9114)https://github.com/apache/seatunnel/commit/ab7119e5072.3.11
[Feature][Checkpoint] Add check script for source/sink state class serialVersionUID missing (#9118)https://github.com/apache/seatunnel/commit/4f5adeb1c72.3.11
[improve] milvus options (#9165)https://github.com/apache/seatunnel/commit/5247e176402.3.11
[Fix][Connector-V2] Fix load state check in MilvusSourceReader to consider partition-level status (#8937)https://github.com/apache/seatunnel/commit/bde235090b2.3.10
[Improve][dist]add shade check rule (#8136)https://github.com/apache/seatunnel/commit/51ef8000162.3.9
[Improve][Core] Refactor common options of column/row (#7911)https://github.com/apache/seatunnel/commit/d1582afee62.3.9
[Feature][connector-milvus] update milvus connector to support dynamic schema, failed retry, etc. (#7885)https://github.com/apache/seatunnel/commit/6a31f917292.3.9
[Feature][Restapi] Allow metrics information to be associated to logical plan nodes (#7786)https://github.com/apache/seatunnel/commit/6b7c53d03c2.3.9
[Fix][Connector-V2] Fix known directory create and delete ignore issues (#7700)https://github.com/apache/seatunnel/commit/e2fb6795772.3.8
[Improve][Connector-V2] Optimize milvus code (#7691)https://github.com/apache/seatunnel/commit/1eddb8e1b12.3.8
[Improve][Connector-V2] Optimize milvus-connector config code (#7658)https://github.com/apache/seatunnel/commit/f831f7a5ec2.3.8
[Improve][Connector-V2] update vectorType (#7446)https://github.com/apache/seatunnel/commit/1bba72385b2.3.8
[Improve][API] Move catalog open to SaveModeHandler (#7439)https://github.com/apache/seatunnel/commit/8c2c5c79a12.3.8
[Feature][Connector-V2] Fake Source support produce vector data (#7401)https://github.com/apache/seatunnel/commit/6937d10ac32.3.8
[Feature][Connector-V2][Milvus] Support Milvus source & sink (#7158)https://github.com/apache/seatunnel/commit/0c69b9166e2.3.6