Mise en cache des résultats de vectorisation - Realtime Compute for Apache Flink

Informations générales

La fonction AI_EMBED utilise le paramètre CACHE_TABLE pour désigner une table externe servant de cache persistant pour les résultats de vectorisation. Son fonctionnement à l'exécution est le suivant :

Hit de cache : si la recherche asynchrone trouve une correspondance, la fonction renvoie directement le résultat sans appeler le modèle.
Miss de cache : si la recherche asynchrone ne trouve aucune correspondance, la fonction appelle le modèle pour calculer le vecteur, renvoie le résultat, puis écrit ce dernier de manière asynchrone dans la table de cache.

Ce mécanisme évite les appels redondants au modèle lorsque plusieurs requêtes utilisent le même texte d'entrée.

Limites

Cette fonctionnalité est prise en charge uniquement par Realtime Compute for Apache Flink VVR 11.8.preview1 et versions ultérieures.
Activer Stream Storage Apache Fluss Edition. La table de cache doit être une table à clé primaire Fluss.

Créer une table de cache

Le type de la clé primaire de la table de cache doit être compatible avec le type du paramètre CACHE_KEY.

Mode avec clé spécifiée explicitement

Le type de la clé primaire doit être compatible avec le type de la colonne indiquée par CACHE_KEY. Par exemple, si la colonne cache_key de la table source est de type INT :

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

Mode avec clé générée automatiquement

La clé primaire doit être de type STRING pour stocker la valeur de hachage SHA256 :

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

Syntaxe

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[...]]
)

Paramètres

Paramètre

Obligatoire

Description

MODEL

Oui

Référence vers le modèle. Pour plus d'informations, consultez la section Enregistrer un modèle.

INPUT

Oui

Colonne contenant le texte d'entrée.

CACHE_TABLE

Non

Référence vers la table de cache. La syntaxe est TABLE <catalog>.<db>.<table>.

CACHE_KEY

Non

Colonne utilisée comme clé de cache. La syntaxe est DESCRIPTOR(<column>). Si ce paramètre n'est pas spécifié, le système génère automatiquement une clé composite à l'aide d'un hachage SHA256.

CONFIG

Non

Configuration d'exécution.

Modes de CACHE_KEY

Spécification explicite (recommandée) : utilise la valeur d'une colonne spécifique de la table d'entrée comme clé de recherche. Ce mode est recommandé lorsque vos données possèdent déjà une clé unique.

CACHE_KEY => DESCRIPTOR(cache_key)

Génération automatique : si vous ne transmettez pas le paramètre CACHE_KEY, le système génère une clé selon les règles suivantes :

-- 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`

Ce mode entraîne une surcharge liée au calcul du hachage SHA256 à l'exécution, mais il garantit automatiquement l'unicité de la clé pour chaque combinaison de modèle, de dimension et de texte d'entrée.

Optimisation des performances

Paramètres d'optimisation

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';

Description des paramètres

Paramètre	Valeur par défaut	Description
`max-concurrent-operations`	10	Nombre maximal d'enregistrements traités simultanément lors des recherches dans le cache ou des appels au modèle. L'augmentation de cette valeur peut améliorer le débit. Toutefois, une valeur trop élevée risque de surcharger le service de modèle ou d'entraîner une utilisation excessive de la mémoire.
`output-mode`	`ORDERED`	Mode de tri des résultats en sortie. `ORDERED` : garantit que les résultats sont produits dans le même ordre que les entrées. Ce mode convient aux opérations en aval dépendant de l'ordre, telles que les jointures ou les fonctions de fenêtrage. `ALLOW_UNORDERED` : produit les enregistrements dès qu'ils sont prêts. Ce mode est recommandé pour les flux en insertion seule.
`timeout`	3 min	Durée maximale autorisée pour la recherche et la prédiction combinées. Si cette valeur est définie trop bas, une opération normale lors d'un miss de cache peut expirer.

Recommandation : l'utilisation du mode ALLOW_UNORDERED offre des avantages significatifs en termes de performances pour les raisons suivantes :

Un hit de cache nécessite généralement une seule recherche, tandis qu'un miss de cache implique un appel supplémentaire au modèle.
En mode ORDERED, un seul miss de cache bloque la production de tous les hits de cache suivants.
En mode ALLOW_UNORDERED, les hits de cache sont produits immédiatement sans attendre la fin des misses précédents.

Remarque

Seuls les flux en insertion seule utilisent le mode ALLOW_UNORDERED. Pour les flux comportant des rétractations, le système revient automatiquement au mode ORDERED, même si le mode ALLOW_UNORDERED est spécifié.

Métriques de surveillance

Lors de l'exécution d'un job, vous pouvez utiliser les métriques Flink pour surveiller l'efficacité du cache.

Nom de la métrique

Description

Notes

ai_function_cache.cache_hit

Nombre de hits de cache

Taux de hit = hits / (hits + misses).

ai_function_cache.cache_miss

Nombre de misses de cache

Une valeur constamment élevée suggère que le cache n'est pas préchargé ou que les clés sont instables.

ai_function_cache.writeback_attempt

Nombre de tentatives d'écriture en retour

Cette valeur doit être proche du nombre de misses.

ai_function_cache.writeback_skipped_null_output

Écritures en retour ignorées en raison d'une sortie nulle

Une valeur élevée indique un problème potentiel avec le modèle.

Exemple

Exemple utilisant une table à clé primaire 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
);

Cet exemple utilise une table à clé primaire Fluss comme cache et définit le mode de sortie sur ALLOW_UNORDERED pour améliorer le débit.

-- 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`
));