VECTOR_SEARCH - Realtime Compute for Apache Flink: Central de Documentação do Alibaba Cloud

A função VECTOR_SEARCH localiza os itens semanticamente mais semelhantes com base em um vetor numérico de alta dimensão especificado. Ela consulta uma tabela de vetores do Milvus a partir de um job Flink SQL e retorna as K entradas mais similares (top-K).

Limitações

Suporte a versões: O Ververica Runtime (VVR) 11.3 e posteriores oferecem suporte ao modo stream. O VVR 11.4 e posteriores oferecem suporte ao modo batch.
Tabela de vetores: Apenas o Milvus é compatível como tabela de vetores.
Tipo de stream: Somente streams do tipo insert-only são aceitas (streams que contêm apenas mensagens INSERT).
Modo de execução: A função VECTOR_SEARCH opera exclusivamente no modo stream; não há suporte para o modo batch.

Sintaxe

VECTOR_SEARCH(
  TABLE <SEARCH_TABLE>,
  DESCRIPTOR(<COLUMN_TO_SEARCH>),
  <COLUMN_TO_QUERY>,
  <TOP_K>[,
  <CONFIG>]
)

Parâmetros

Parâmetro	Tipo	Descrição
`TABLE <SEARCH_TABLE>`	TABLE	Nome da tabela de vetores.
`DESCRIPTOR(<COLUMN_TO_SEARCH>)`	DESC	Coluna de vetor indexada na tabela de vetores. O sistema compara os dados de entrada com esta coluna para calcular a similaridade.
`COLUMN_TO_QUERY`	`ARRAY<FLOAT>` / `ARRAY<DOUBLE>`	Coluna de embedding do stream de entrada, como o embedding de uma imagem ou texto carregado.
`TOP_K`	INT	Quantidade máxima de entradas similares retornadas por linha de entrada.
`CONFIG`	`MAP<STRING,STRING>`	Parâmetros opcionais de tempo de execução. Consulte Parâmetros de tempo de execução.

Valor de retorno

A função VECTOR_SEARCH retorna uma tabela. Cada linha contém todas as colunas da tabela de vetores, além de uma coluna score do tipo DOUBLE. Essa coluna score indica a similaridade entre os dados de entrada e os dados de saída.

Parâmetros de tempo de execução

Transmita os parâmetros de tempo de execução como um MAP<STRING,STRING> no argumento CONFIG, por exemplo: MAP['async', 'false'].

Por padrão, o mecanismo seleciona o modo de execução conforme o suporte do conector Milvus. Caso o conector aceite tanto o modo assíncrono quanto o síncrono, o mecanismo prioriza o modo assíncrono para maximizar o throughput. Defina explicitamente o parâmetro async para substituir esse comportamento.

Parâmetro	Tipo	Padrão	Descrição
`async`	Boolean	(nenhum)	Define se o modo assíncrono deve ser utilizado. Se o conector não oferecer suporte ao modo especificado, o mecanismo reportará um erro.
`max-concurrent-operations`	Integer	`10`	Número máximo de requisições simultâneas permitidas no modo assíncrono.
`output-mode`	Enum	`ORDERED`	Modo de saída para operações assíncronas. Valores válidos: `ORDERED`, `ALLOW_UNORDERED`. Para mais detalhes, consulte Async I/O — Order of results.
`timeout`	Duration	`3 min`	Tempo limite para uma operação assíncrona, contado desde a primeira chamada até a conclusão. Esse período pode incluir múltiplas tentativas e é reiniciado em caso de failover.

Exemplo

Dados de teste

vector_table:

id	topic	vector_index
1	"BigData"	[1, 1, 0]
2	"Streaming"	[-5, -12, -13]
3	"Batch"	[5, 12, 13]

query_table:

id	user_keyword	embedding
1	"Spark"	[5, 12, 13]
2	"Flink"	[-5, -12, -13]

Consulta

A instrução abaixo utiliza cada linha da query_table para pesquisar na vector_table e recuperar os dois registros mais semelhantes. A opção MAP['async', 'false'] ativa explicitamente o modo síncrono.

SELECT user_keyword, topic
FROM
  query_table,
  LATERAL TABLE (VECTOR_SEARCH(
    SEARCH_TABLE => TABLE vector_table,
    COLUMN_TO_SEARCH => DESCRIPTOR(vector_index),
    COLUMN_TO_QUERY => query_table.embedding,
    TOP_K => 2,
    MAP['async', 'false'] -- Enable synchronous mode
    ))

Resultados

user_keyword	topic
"Spark"	"Batch"
"Spark"	"BigData"
"Flink"	"Streaming"
"Flink"	"BigData"