AI_EXTRACT uses Flink operators with Large Language Model (LLM) services to extract structured information. It requires Ververica Runtime (VVR) 11.4 or later and is subject to rate limiting by the model platform. - Realtime Compute for Apache Flink

AI_EXTRACT extrait des informations structurées du texte brut à l'aide d'un grand modèle de langage (LLM).

Prérequis

Avant d'utiliser AI_EXTRACT, enregistrez un service de modèle. Pour plus de détails, consultez la rubrique Paramètres du modèle.

Le type de sortie du modèle doit être VARIANT.

Syntaxe

AI_EXTRACT(
  MODEL => MODEL <MODEL NAME>,
  INPUT => <INPUT COLUMN NAME>,
  EXTRACT_SCHEMA => <EXTRACT SCHEMA>
)

AI_EXTRACT prend en charge les arguments nommés (illustrés ci-dessus) et les arguments positionnels. Consultez l'Exemple pour voir les deux styles d'appel.

Paramètres

Paramètre	Type de données	Description
`MODEL <MODEL NAME>`	MODEL	Nom du service de modèle enregistré.
`<INPUT COLUMN NAME>`	STRING	Colonne contenant le texte brut dont il faut extraire les informations.
`<EXTRACT_SCHEMA>`	STRING	Chaîne JSON définissant les champs à extraire et leurs types de données. Doit être une constante.

Format de EXTRACT_SCHEMA

EXTRACT_SCHEMA accepte un objet JSON où chaque clé correspond au nom du champ à extraire et chaque valeur indique le type de données cible :

{"<field_name>": "<data_type>", ...}

Exemple de schéma :

{"name": "string", "age": "int", "address": "string", "phone": "string"}

Sortie

Colonne	Type de données	Description
`extracted_json`	STRING	Champs extraits sous forme de chaîne JSON.

Exemple

Cet exemple extrait des informations personnelles (nom, âge, adresse et numéro de téléphone) d'une colonne de texte à l'aide d'un modèle intégré à Flink.

Étape 1 : Enregistrer le modèle

CREATE TEMPORARY MODEL general_model
INPUT (`input` STRING)
OUTPUT (`content` VARIANT)
WITH (
    'provider' = 'openai-compat',
    'task'     = 'chat/completions',
    'model'    = 'qwen3.6-flash'
);

Étape 2 : Appeler AI_EXTRACT

CREATE TEMPORARY VIEW infos(id, description)
AS VALUES (1, 'Xiao Ming is 18 years old and lives in Hangzhou. His phone number is ******.');

-- Named argument style
SELECT id, extracted_json
FROM infos,
LATERAL TABLE(
  AI_EXTRACT(
    MODEL => MODEL general_model,
    INPUT => description,
    EXTRACT_SCHEMA => '{"name":"string","phone":"string","address":"string","age":"int"}'));

-- Positional argument style
SELECT id, extracted_json
FROM infos,
LATERAL TABLE(
  AI_EXTRACT(
    MODEL general_model,
    description,
    '{"name":"string","phone":"string","address":"string","age":"int"}'));

Sortie :

id	extracted_json
1	`{"address":"Hangzhou","age":18,"name":"Xiao Ming","phone":"******"}`

Limites

Pris en charge uniquement par le moteur de calcul en temps réel Ververica Runtime (VVR) 11.4 ou version ultérieure.
Pour utiliser le service Flink AI (modèles intégrés), vous devez disposer de VVR 11.7 ou version ultérieure et activer le service Flink AI. Pour plus de détails, consultez la rubrique Service Flink AI (modèles intégrés).
Le débit est limité par la plateforme de service de modèle. Lorsque le trafic atteint la limite d'accès de la plateforme, une contre-pression se produit sur les opérateurs Flink utilisés dans les instructions AI_EXTRACT. Une limitation sévère du débit peut provoquer des délais d'expiration et entraîner le redémarrage du job Flink.