pyflink.dataframe.DataFrame.write_milvus#

DataFrame.write_milvus(endpoint: str, *, username: str, password: str, database_name: str, collection_name: str, primary_key: str, port: int = 19530, partition_name: str = '_default', partition_key_enabled: bool = False, max_retries: int = 3, ignore_delete: bool = False, parallelism: Optional[int] = None, buffer_flush_max_rows: int = 200, buffer_flush_interval_ms: int = 3000, extra_options: Optional[Dict[str, Any]] = None) → None[source]#

Write the DataFrame to a Milvus collection.

Milvus is a vector database. This sink supports upsert semantics: rows with the same primary key replace previous values. The underlying Java connector requires exactly one primary key column, so primary_key accepts a single column name.

The DataFrame’s current schema is forwarded to the sink, with the primary_key column marked NOT NULL and declared as the table primary key. Make sure your DataFrame already contains a column with that name.

Parameters

endpoint – Milvus service endpoint URL, e.g. "http://localhost" or "https://my-instance.api.gcp-us-west1.zillizcloud.com".
username – Milvus user name for authentication.
password – Milvus password for authentication.
database_name – Target database name.
collection_name – Target collection name.
primary_key – Name of the single primary key column. The column must exist in the DataFrame schema.
port – Milvus service port. Defaults to 19530. Must be in [0, 65535].
partition_name – Target partition name. Defaults to "_default".
partition_key_enabled – Whether the collection has the partition-key feature enabled. Defaults to False.
max_retries – Maximum number of retries when writing to Milvus fails. Defaults to 3. Must be >= 0.
ignore_delete – Whether to ignore delete messages from the input stream. Defaults to False.
parallelism – Optional sink parallelism. If set, must be >= 1.
buffer_flush_max_rows – Maximum number of buffered rows before a flush is triggered. Defaults to 200. Must be >= 0.
buffer_flush_interval_ms – Maximum buffer time in milliseconds before a flush is triggered. Defaults to 3000. Must be >= 0.
extra_options – Additional connector options. Values override named options except "connector", which is reserved.

Example:

>>> import pyflink.dataframe as pf
>>> df = pf.from_records(
...     [(1, [0.1, 0.2]), (2, [0.3, 0.4])],
...     schema=["id", "vector"],
... )
>>> df.write_milvus(
...     endpoint="http://milvus.example.com",
...     username="${secret_values.milvus_user}",
...     password="${secret_values.milvus_password}",
...     database_name="my_db",
...     collection_name="my_collection",
...     primary_key="id",
... )