Realtime Compute for Apache Flink:Milvus connector (public preview)

Overview

Milvus is a highly scalable vector database designed to handle large-scale unstructured data, such as images, text, and audio. It supports efficient similarity search, making it ideal for use cases such as recommendation systems, image retrieval, and semantic search. The Milvus connector supports the following capabilities:

Features

The Milvus connector tightly integrates Apache Flink with the Milvus vector database to create a high-performance, reliable data pipeline for real-time vector search scenarios. Key features of the Milvus connector include:

• High-concurrency writes: Supports configurable sink parallelism.

• Automatic retry: Retries failed operations to improve stability.

• Batched buffering: Improves write performance by batching records before they are flushed.

• At-least-once semantics: Ensures eventual consistency through idempotent updates based on the primary key.

• Vector search: Enables real-time vector similarity searches directly within Flink SQL.

Prerequisites

• You have created a Milvus cluster. For more information, see Quickly create a Milvus instance.

• You have created a Milvus collection. If you plan to write to a specific partition, ensure that the Milvus partition already exists.

Limitations

• Writing to a sink table requires Ververica Runtime (VVR) 11.1 or later.

• Querying a vector table requires VVR 11.3 or later.

• Only Milvus 2.4.x is supported.

• The Milvus connector only supports at-least-once semantics.

Syntax

CREATE TEMPORARY TABLE milvus_sink (
  id BIGINT,
  f1 STRING,
  f2 BOOLEAN,
  f3 TINYINT,
  f4 SMALLINT,
  f5 INTEGER,
  f6 DATE,
  f7 TIME(3),
  f8 TIMESTAMP_LTZ(3),
  f9 TIMESTAMP(3),
  f10 FLOAT,
  f11 DOUBLE,
  f12 DECIMAL(10, 2),
  f13 ARRAY<FLOAT>,
  f14 ARRAY<DOUBLE>,
  f15 ARRAY<INTEGER>,
  f16 ARRAY<BIGINT>,
  PRIMARY KEY (id) NOT ENFORCED  -- Required. Milvus supports only BIGINT or STRING as the primary key.
) WITH (
  'connector'='milvus',
  'endpoint'='<yourEndpoint>',
  'port'='<yourPort>',
  'userName'='<yourUserName>',
  'password'='<yourPassword>',
  'databaseName'='<yourDatabaseName>',
  'collectionName'='<yourCollectionName>'
);

Connector options

Type mapping

Examples

• Write streaming data to Milvus

-- Create a mock data source that generates 100 rows per second. simulated streaming data.
CREATE TEMPORARY TABLE mock_source (
    id STRING,
    vector ARRAY<FLOAT>,       -- Vector passed as a FLOAT array.
    event_time AS PROCTIME() -- Processing time attribute.
) WITH (
    'connector' = 'datagen',
    'rows-per-second' = '100',  -- Generates 100 rows per second.
    'fields.id.kind' = 'sequence',
    'fields.id.start' = '1',
    'fields.id.end' = '1000'
);

CREATE TEMPORARY TABLE milvus_sink (
  id STRING,               -- Unique identifier, such as a device ID.
  vector ARRAY<FLOAT>,     -- Vector data. The array length must be consistent with the source.
  timestamp BIGINT         -- Timestamp for stream processing.
  PRIMARY KEY (id) NOT ENFORCED  -- Required. Milvus supports only BIGINT or STRING as the primary key.
) WITH (
  'connector'='milvus',
  'endpoint'='xxx',
  'port'='19530',
  'userName'='xxx',
  'password'='xxx',
  'databaseName'='xxxx',
  'collectionName'='xxxx'
);

-- Transform data and write to Milvus.
INSERT INTO milvus_sink
SELECT 
    id,
    vector,
    UNIX_TIMESTAMP() * 1000 AS timestamp  -- Current timestamp in milliseconds.
FROM mock_source;

• Perform a vector search

CREATE TEMPORARY TABLE milvus_table (
  id STRING,               -- Unique identifier.
  vector ARRAY<FLOAT>,     -- Vector data. The array length must be consistent with the source.
  PRIMARY KEY (id) NOT ENFORCED  -- Required. Milvus only supports BIGINT or STRING as the primary key.
) WITH (
  'connector'='milvus',
  'endpoint'='xxx',
  'port'='19530',
  'userName'='xxx',
  'password'='xxx',
  'databaseName'='xxxx',
  'collectionName'='xxxx'
);
-- Find the top 2 most similar items for the vector [1.1, 2.2, 3.3].
SELECT * FROM 
LATERAL TABLE(
  VECTOR_SEARCH(
    TABLE milvus_table, 
    DESCRIPTOR(vector), 
    ARRAY[1.1, 2.2, 3.3], 
    2));

Note: The Milvus collection must be loaded into memory before running the query. For more information, see the Milvus documentation on Load a collection.