pyflink.dataframe.read_milvus#

read_milvus(endpoint: str, *, schema: Dict[str, DataType], computed_columns: Optional[Dict[str, str]] = None, watermark: Optional[Tuple[str, str]] = None, username: str, password: str, database_name: str, collection_name: str, port: int = 19530, partition_name: str = '_default', partition_key_enabled: bool = False, max_retries: int = 3, search_metric: Literal['L2', 'IP', 'COSINE'] = 'L2', extra_options: Optional[Dict[str, Any]] = None) → pyflink.dataframe.dataframe.DataFrame[source]#

Read a Milvus vector search source into a DataFrame.

Parameters

endpoint – Milvus service endpoint URL.
schema – Dict of {column_name: DataType} describing the source schema. Required and must not be empty.
computed_columns – Optional dict of computed column expressions keyed by computed column name.
watermark – Optional tuple of (rowtime_column, watermark_expression).
username – Milvus user name for authentication.
password – Milvus password for authentication.
database_name – Source database name.
collection_name – Source collection name.
port – Milvus service port. Defaults to 19530.
partition_name – Source partition name. Defaults to "_default".
partition_key_enabled – Whether partition-key routing is enabled.
max_retries – Maximum retry count for connector operations.
search_metric – Vector search metric, one of "L2", "IP", or "COSINE".
extra_options – Additional connector options forwarded through to the underlying Milvus connector. This is for options not exposed as named parameters of read_milvus. Keys matching options exposed as named parameters are rejected; use the named parameters instead. "connector" is reserved and must not be supplied.

Returns

A new DataFrame backed by the configured Milvus source.

Example:

>>> import pyflink.dataframe as pf
>>>
>>> vectors = pf.read_milvus(
...     "http://milvus.example.com",
...     schema={
...         "id": pf.DataType.int64(),
...         "ts_millis": pf.DataType.int64(),
...     },
...     computed_columns={
...         "event_time": "TO_TIMESTAMP_LTZ(ts_millis, 3)",
...     },
...     watermark=("event_time", "event_time - INTERVAL '5' SECOND"),
...     username="root",
...     password="${secret_values.milvus_password}",
...     database_name="default",
...     collection_name="items",
... )