Milvus

The milvus output upserts each batch's rows into a Milvus collection over the REST v2 vectordb API (Milvus 2.4+). It pairs with the embedding processor: a Float32 list column becomes the vector, every remaining column packs into a JSON payload field, and an optional id column keys the row.

Configuration

Field	Type	Required	Default	Description
type	string	yes	—	milvus
url	string	yes	—	Milvus base URL (e.g. http://localhost:19530).
collection	string	yes	—	Target collection (schema must already exist).
vector_field	string	no	embedding	Field carrying the vector (FixedSizeList/List of Float32).
id_field	string	yes	—	Field carrying the row id (integer or string). Required: the REST v2 upsert API needs a primary key on every row, so auto-id collections are not supported (the builder rejects a missing id_field).
payload_field	string	no	payload	JSON field receiving every remaining column as a per-row object. Set to "" to disable.
api_key	string	no	—	Sent as Authorization: Bearer; Milvus convention is <user>:<password>. Supports secret references.
timeout_ms	integer	no	30000	HTTP request timeout.
retry_count	integer	no	0	Retry attempts for connection errors and 5xx responses (exponential backoff starting at 100ms). Batches above 1000 rows are split into multiple requests regardless.
headers	map	no	—	Extra HTTP headers.

Collection schema

Create the collection before running; the payload field must be a JSON type (or enable the dynamic field and keep payload_field on the dynamic key):

bash

# via curl, once:
curl -s http://localhost:19530/v2/vectordb/collections/create -d '{
  "collectionName": "documents",
  "dimension": 1536
}'

Each batch of up to 1,000 rows produces one upsert request; larger batches are split into sequential 1,000-row requests:

json

{"collectionName": "documents", "data": [
  {"doc_id": 1, "embedding": [0.1, "..."], "payload": {"text": "..."}}
]}

note

Milvus reports failures as HTTP 200 with a non-zero code in the body — this output treats that as an error and surfaces code plus message. With id_field set, redelivery overwrites the same row (at-least-once friendly); without it every attempt inserts a new row (auto-id). Requests to loopback endpoints always bypass the system proxy.

Example

yaml

output:
  type: "milvus"
  url: "http://localhost:19530"
  collection: "documents"
  vector_field: "embedding"
  id_field: "doc_id"
  api_key: "${env:MILVUS_CREDENTIALS:-}"

Complete Pipeline Example

yaml

logging:
  level: info

streams:
  - id: docs-to-milvus
    input:
      type: memory
    pipeline:
      thread_num: 4
      processors:
        - type: json_to_arrow
        - type: embedding
          api_base: "http://localhost:9997/v1"
          model: "text-embedding-3-small"
          api_key: "${env:OPENAI_API_KEY:-}"
          field: "text"
    output:
      type: milvus
      url: "http://localhost:19530"
      collection: "documents"
      vector_field: "embedding"
      id_field: "doc_id"