Caché de resultados de vectorización - Realtime Compute for Apache Flink

Este documento describe cómo configurar una tabla de caché externa para la función AI_EMBED. De esta forma, se reutilizan los resultados de vectorización cuando el texto de entrada es idéntico. Esto reduce las llamadas al modelo, disminuye la latencia y ahorra costos.

Contexto

La función AI_EMBED utiliza el parámetro CACHE_TABLE para especificar una tabla externa como caché persistente de los resultados de vectorización. Durante la ejecución, funciona de la siguiente manera:

Acierto de caché: si la búsqueda asíncrona encuentra una coincidencia, la función devuelve el resultado directamente y omite la llamada al modelo.
Fallo de caché: si la búsqueda asíncrona no encuentra ninguna coincidencia, la función llama al modelo para calcular el vector y devuelve el resultado. Después, lo escribe de forma asíncrona en la tabla de caché.

Este mecanismo evita llamadas redundantes al modelo cuando se repiten solicitudes con el mismo texto de entrada.

Límites

Esta función solo es compatible con Realtime Compute for Apache Flink VVR 11.8.preview1 y versiones posteriores.
Enable Stream Storage Apache Fluss Edition. La tabla de caché debe ser una tabla de clave primaria de Fluss.

Crear una tabla de caché

El tipo de la clave primaria de la tabla de caché debe ser compatible con el tipo de CACHE_KEY.

Modo de clave especificada de forma explícita

El tipo de la clave primaria debe ser compatible con el tipo de la columna que especifica CACHE_KEY. Por ejemplo, si cache_key en la tabla de origen es de tipo INT:

CREATE TABLE `fluss`.`default`.`ai_cache` (
  cache_key   INT NOT NULL,
  embedding   ARRAY<FLOAT>,PRIMARY KEY (cache_key) NOT ENFORCED
);

Modo de clave generada automáticamente

La clave primaria debe ser de tipo STRING para almacenar el valor de hash SHA256:

CREATE TABLE `fluss`.`default`.`ai_cache_sha` (
  cache_key   STRING NOT NULL,
  embedding   ARRAY<FLOAT>,PRIMARY KEY (cache_key) NOT ENFORCED
);

Sintaxis

AI_EMBED(
  MODEL       => MODEL <MODEL NAME>,
  INPUT       => <INPUT COLUMN NAME>
  [, DIMENSION   => <DIMENSION VALUE>]
  [, CACHE_TABLE => TABLE <catalog>.<db>.<table>
     [, CACHE_KEY => DESCRIPTOR(<column>)]]
  [, CONFIG      => MAP[...]]
)

Parámetros

Parámetro	Obligatorio	Descripción
`MODEL`	Sí	Referencia al modelo. Para obtener más información, consulte Register model.
`INPUT`	Sí	Columna de texto de entrada.
`CACHE_TABLE`	No	Referencia a la tabla de caché. La sintaxis es `TABLE <catalog>.<db>.<table>`.
`CACHE_KEY`	No	Columna de la clave de caché. La sintaxis es `DESCRIPTOR(<column>)`. Si no se especifica, el sistema genera automáticamente una clave compuesta mediante un hash SHA256.
`CONFIG`	No	Configuración de tiempo de ejecución.

Modos de CACHE_KEY

Especificación explícita (recomendado): utiliza el valor de una columna específica de la tabla de entrada como clave de búsqueda. Se recomienda este modo cuando los datos ya cuentan con una clave única.

CACHE_KEY => DESCRIPTOR(cache_key)

Generación automática: si no pasa CACHE_KEY, el sistema genera una clave según las siguientes reglas:

-- If CACHE_KEY is not provided, the system automatically generates the key:
-- key = LOWER(SHA256(model_id || ':' || dimension || ':' || input_text))
CACHE_TABLE => TABLE `fluss-ql`.`default`.`ai_cache1`

Este modo implica la sobrecarga de calcular un hash SHA256 durante la ejecución. Sin embargo, garantiza automáticamente una clave única para cada combinación de modelo, dimensión y texto de entrada.

Ajuste de rendimiento

Parámetros de ajuste

SET 'table.exec.async-ml-predict.max-concurrent-operations' = '100';
SET 'table.exec.async-ml-predict.output-mode' = 'ALLOW_UNORDERED';
SET 'table.exec.async-ml-predict.timeout' = '120s';

Descripción de los parámetros

Parámetro	Predeterminado	Descripción
`max-concurrent-operations`	10	Número máximo de registros para búsquedas de caché o llamadas al modelo simultáneas. Aumentar este valor mejora el throughput. No obstante, establecerlo demasiado alto puede sobrecargar el servicio del modelo o provocar un uso excesivo de memoria.
`output-mode`	`ORDERED`	Modo de orden de salida. `ORDERED`: garantiza que la salida mantenga el mismo orden que la entrada. Es adecuado para operaciones posteriores que dependen del orden, como las uniones o las funciones de ventana. `ALLOW_UNORDERED`: emite los registros en cuanto se completan. Se recomienda para flujos de solo inserción.
`timeout`	3min	Tiempo total máximo permitido tanto para la búsqueda como para la predicción. Si este valor es demasiado bajo, una operación normal ante un fallo de caché podría agotar el tiempo de espera.

Recomendación: utilizar ALLOW_UNORDERED ofrece ventajas de rendimiento significativas por las siguientes razones:

Un acierto de caché suele requerir solo una búsqueda, mientras que un fallo necesita una llamada adicional al modelo.
En el modo ORDERED, un solo fallo de caché bloquea la salida de todos los aciertos posteriores.
En el modo ALLOW_UNORDERED, los aciertos de caché se emiten de inmediato, sin esperar a que finalicen los fallos anteriores.

Nota

Solo los flujos de solo inserción utilizan el modo ALLOW_UNORDERED. En los flujos con retractaciones, el sistema vuelve al modo ORDERED, incluso si se especifica ALLOW_UNORDERED.

Métricas de monitoreo

Mientras un trabajo está en ejecución, puede usar las métricas de Flink para monitorear la eficacia de la caché.

Nombre de la métrica	Descripción	Notas
`ai_function_cache.cache_hit`	recuento de aciertos de caché	Tasa de aciertos = aciertos / (aciertos + fallos).
`ai_function_cache.cache_miss`	recuento de fallos de caché	Un valor constantemente alto indica que la caché no se ha precalentado o que las claves son inestables.
`ai_function_cache.writeback_attempt`	recuento de intentos de escritura diferida	Este valor debería ser cercano al recuento de fallos.
`ai_function_cache.writeback_skipped_null_output`	escrituras diferidas omitidas por una salida nula	Un valor alto señala un posible problema con el modelo.

Ejemplo

A continuación se muestra un ejemplo de una tabla de clave primaria de Fluss:

-- Create the cache table (Fluss) in advance
CREATE TABLE `fluss`.`default`.`embedding_cache` (
  cache_key   STRING NOT NULL,
  embedding   ARRAY<FLOAT>,
  PRIMARY KEY (cache_key) NOT ENFORCED
);

Este ejemplo usa una tabla de clave primaria de Fluss como caché y establece el modo de salida en ALLOW_UNORDERED para mejorar el throughput.

-- Tuning parameters
SET 'table.exec.async-ml-predict.max-concurrent-operations' = '100';
SET 'table.exec.async-ml-predict.output-mode' = 'ALLOW_UNORDERED';

-- Create the model
CREATE TEMPORARY MODEL embedding_model
  INPUT (text STRING)
  OUTPUT (embedding ARRAY<FLOAT>)
WITH (
  'provider' = 'openai-compat',
  'task' = 'embeddings',
  'model' = 'text-embedding-v4',
  'dimension' = '1024'
);

-- Business query
INSERT INTO result_sink
SELECT id, content, embedding
FROM source_table, LATERAL TABLE(AI_EMBED(
  MODEL       => MODEL embedding_model,
  INPUT       => content,
  CONFIG      => MAP['async', 'true'],
  CACHE_TABLE => TABLE `fluss`.`default`.`embedding_cache`
));