Realtime Compute for Apache Flink:Appel général

CREATE TEMPORARY TABLE text_source (
  user_input STRING
) WITH ('connector' = 'datagen');

CREATE TEMPORARY TABLE result_sink (
  user_input STRING,
  ai_analysis STRING
) WITH ('connector' = 'print');

CREATE TEMPORARY MODEL text_model
INPUT (user_input STRING)
OUTPUT (content STRING)
WITH (
  'provider' = 'openai-compat',
  'model' = 'qwen3.6-flash',
  'task' = 'chat/completions',
  'system-prompt' = 'Rate the gibberish level of the input on a scale of 0 to 100'
);

INSERT INTO result_sink
SELECT user_input, content as ai_analysis FROM
    ML_PREDICT(
        TABLE text_source,
        MODEL text_model,
        DESCRIPTOR(user_input)
);

ML_PREDICT(TABLE <table_name>, MODEL <model_name>, DESCRIPTOR(<input_columns>) [, CONFIG => MAP[...]])        

Paramètres de type de contenu

• Pour une entrée sur une seule colonne, utilisez content-type afin de spécifier le type de contenu. Valeurs prises en charge : text, image_url.

• Pour une entrée multicolumnes, utilisez content-types afin de spécifier le type de contenu pour chaque colonne. Les valeurs prises en charge et leur correspondance avec les types de colonnes Flink SQL sont listées ci-dessous :

• Il est recommandé de spécifier content-types dans les paramètres. Si vous définissez à la fois content-type et content-types, le framework renvoie une erreur. Remarque : les paramètres content-type et content-types définis dans les paramètres du modèle et dans la configuration par appel s'influencent mutuellement :

Paramètres au niveau des colonnes

Traitement par lots (Bundle processing)

• Traitement : Le traitement par lots agrège d'abord plusieurs enregistrements d'entrée en un lot, puis appelle le modèle d'IA pour ce lot. Cela réduit le nombre d'appels au modèle, mais augmente le temps d'inférence pour chaque requête. Lorsque les enregistrements peuvent être efficacement agrégés, le traitement par lots améliore considérablement le débit de l'opérateur.

• Description des paramètres : bundle-size spécifie le nombre maximal d'enregistrements accumulés dans un lot. bundle-allow-latency spécifie le temps d'attente maximal lorsqu'un lot n'a pas atteint la valeur bundle-size. Le traitement démarre dès que l'une des conditions (taille du lot ou temps d'attente) est remplie.

• Exigences : Le traitement par lots prend effet en fonction du schéma d'entrée et de sortie du modèle.

  • Génération de texte : Pour les modèles dont la task est chat/completions, l'entrée contient exactement une colonne de texte de type STRING et la sortie contient exactement une colonne de type STRING ou VARIANT. Les entrées multicolumnes et les entrées multimodales telles que les images, les vidéos et l'audio ne sont pas prises en charge.

  • Embedding de texte : Pour les modèles dont la task est embeddings, l'entrée contient exactement une colonne de type STRING et la sortie contient exactement une colonne de type ARRAY<FLOAT>.

• Fonctions prises en charge : ML_PREDICT, classification de texte (AI_CLASSIFY), analyse des sentiments (AI_SENTIMENT), extraction d'informations (AI_EXTRACT), résumé de texte (AI_SUMMARIZE), traduction de texte (AI_TRANSLATE), masquage des données (AI_MASK), et embedding de texte (AI_EMBED). Parmi celles-ci, ML_PREDICT prend en charge le traitement par lots uniquement si le modèle enregistré répond aux exigences précédentes.

• Configuration globale : Cette configuration s'applique aux appels ML_PREDICT et aux appels aux fonctions d'IA spécifiques répertoriées dans la tâche qui satisfont aux exigences précédentes. Utilisez les paramètres globaux suivants pour activer le traitement par lots :

  -- Bundle processing is supported only in VVR 11.9 and later.
  SET 'table.exec.ml-predict.bundle-size' = '10';
  SET 'table.exec.ml-predict.bundle-allow-latency' = '5s';

• Configuration par appel : Ces paramètres s'appliquent uniquement à l'appel de fonction dans lequel ils sont spécifiés. ML_PREDICT et les fonctions d'IA spécifiques répertoriées utilisent la même méthode de configuration. Pour les paramètres et les exemples, consultez la section Configuration par appel.

Limites

• Pris en charge uniquement à partir de la version VVR 11.9.

• Vous ne pouvez pas activer la gestion des erreurs lorsque les appels de modèle s'exécutent en mode synchrone.

• La gestion des erreurs ne peut pas être utilisée conjointement avec le Cache de résultats de vectorisation (CACHE_TABLE) pour le moment.

Activer la gestion des erreurs

Ajoutez le paramètre FAIL_ON_ERROR => FALSE lors de l'appel à une fonction d'IA.

Prérequis : Conservez les paramètres de nouvelle tentative par défaut du modèle : error-handling-strategy='retry' et retry-fallback-strategy='failover'. Pour plus d'informations sur les paramètres du modèle, consultez la page Paramètres du modèle.

Modifications de la sortie

Après avoir activé la gestion des erreurs, le système ajoute une colonne d'erreur err_msg de type VARIANT nullable à la fin des champs de sortie d'origine. Cette colonne est générée par le système. Vous n'avez pas besoin de la déclarer dans la table d'entrée ni dans la clause OUTPUT de CREATE MODEL.

• Si l'appel réussit : err_msg est NULL.

• Si l'appel échoue définitivement : tous les champs de résultat du modèle sont NULL et err_msg contient les informations d'erreur.

Si vous utilisez SELECT *, le résultat inclut la nouvelle colonne d'erreur. Si vous sélectionnez explicitement des champs, vous devez spécifier la colonne manuellement. Le nom de colonne par défaut est err_msg. En cas de conflit de nom avec une colonne de sortie existante, le système ajoute un suffixe numérique, par exemple err_msg0. Utilisez le nom de colonne réel pour faire référence à la colonne.

Exemples

CREATE TEMPORARY VIEW input_data AS
SELECT *
FROM (VALUES
  (1, CAST('xxxx (a description of violent conflict that violates humanity)' AS STRING), 'negative'),
  (2, CAST('The customer service of this store has always been among the best' AS STRING), 'positive')
) AS t(id, input_text, expected_label);

CREATE TEMPORARY VIEW ai_results AS
SELECT i.id, i.input_text, i.expected_label,
       r.score, r.label, r.confidence, r.err_msg
FROM input_data AS i,
LATERAL TABLE(AI_SENTIMENT(
  MODEL => MODEL ai_sentiment_model,
  INPUT => i.input_text,
  FAIL_ON_ERROR => FALSE,
  CONFIG => MAP['async', 'true']
)) AS r(score, label, confidence, err_msg);

SELECT * FROM ai_results;

message: 
java.util.concurrent.TimeoutException: AsyncPredictFunction did not complete within the configured timeout. Increase the ML_PREDICT timeout or check external service connectivity.

stack_trace:
java.util.concurrent.TimeoutException: AsyncPredictFunction did not complete within the configured timeout. Increase the ML_PREDICT timeout or check external service connectivity.
at org.apache.flink.table.runtime.operators.predict.AsyncMLPredictRunner.timeout(AsyncMLPredictRunner.java:153)
at org.apache.flink.table.runtime.operators.predict.AsyncMLPredictRunner.timeout(AsyncMLPredictRunner.java:51)
at org.apache.flink.table.runtime.operators.TableKeyedAsyncWaitOperator$KeyedResultHandler.timerTriggered(TableKeyedAsyncWaitOperator.java:462)
at org.apache.flink.table.runtime.operators.TableKeyedAsyncWaitOperator$KeyedResultHandler.lambda$registerTimeout$1(TableKeyedAsyncWaitOperator.java:457)
at org.apache.flink.table.runtime.operators.TableKeyedAsyncWaitOperator.lambda$registerTimer$4(TableKeyedAsyncWaitOperator.java:486)
at org.apache.flink.streaming.runtime.tasks.StreamTask.invokeProcessingTimeCallback(StreamTask.java:2285)
at org.apache.flink.streaming.runtime.tasks.StreamTask.lambda$deferCallbackToMailbox$28(StreamTask.java:2276)
at org.apache.flink.streaming.runtime.tasks.StreamTaskActionExecutor$1.runThrowing(StreamTaskActionExecutor.java:50)
at org.apache.flink.streaming.runtime.tasks.mailbox.Mail.run(Mail.java:115)
at org.apache.flink.streaming.runtime.tasks.mailbox.MailboxProcessor.runMail(MailboxProcessor.java:415)
at org.apache.flink.streaming.runtime.tasks.mailbox.MailboxProcessor.processMailsWhenDefaultActionUnavailable(MailboxProcessor.java:384)
at org.apache.flink.streaming.runtime.tasks.mailbox.MailboxProcessor.processMail(MailboxProcessor.java:369)
at org.apache.flink.streaming.runtime.tasks.mailbox.MailboxProcessor.runMailboxLoop(MailboxProcessor.java:229)
at org.apache.flink.streaming.runtime.tasks.StreamTask.runMailboxLoop(StreamTask.java:1273)
at org.apache.flink.streaming.runtime.tasks.StreamTask.invoke(StreamTask.java:1217)
at org.apache.flink.runtime.taskmanager.Task.runWithSystemExitMonitoring(Task.java:1133)
at org.apache.flink.runtime.taskmanager.Task.restoreAndInvoke(Task.java:1110)
at org.apache.flink.runtime.taskmanager.Task.doRun(Task.java:914)
at org.apache.flink.runtime.taskmanager.Task.run(Task.java:716)
at java.base/java.lang.Thread.run(Thread.java:991)

Gestion de chaque type d'erreur

Le comportement suivant s'applique lorsque la gestion des erreurs est activée et que les paramètres de nouvelle tentative par défaut sont conservés. Une ligne d'erreur est une ligne qui conserve l'entrée, renvoie des résultats de modèle vides et enregistre la raison de l'échec dans err_msg.

Texte

L'exemple suivant enregistre et utilise un modèle intégré à Flink AI Service pour classer le texte d'entrée selon les sentiments. L'exemple 1 utilise les paramètres au niveau du MODEL. L'exemple 2 remplace user-prompt au moment de l'appel.

-- Register a built-in model. When content-type is not specified, it defaults to text
CREATE MODEL sentiment_model
INPUT (prompt STRING)
OUTPUT (response STRING)
WITH (
  'provider' = 'openai-compat',
  'task' = 'chat/completions',
  'model' = 'qwen3.6-flash',
  'system-prompt' = 'You are a sentiment classifier. Output one label: negative, positive, or neutral'
);

-- Create a source table: simulated product review data
CREATE TEMPORARY VIEW input_table(id, content)
AS VALUES
  (1, 'Great quality, soft fabric, fits perfectly'),
  (2, 'Had loose threads on arrival, faded badly after one wash'),
  (3, 'Received the item, looks as pictured'),
  (4, 'Started pilling after two weeks, customer service refused returns'),
  (5, 'Flattering fit, color is even better than the photo, already ordered a third one');

-- Create a result table
CREATE TEMPORARY TABLE output_table (
  id INT,
  content STRING,
  sentiment STRING
) WITH (
  'connector' = 'print'
);

-- Use ML_PREDICT for real-time inference
-- Example 1: Use MODEL-level parameters
INSERT INTO output_table
SELECT
  id,
  content,
  response AS sentiment
FROM ML_PREDICT(
  TABLE input_table,
  MODEL sentiment_model,
  DESCRIPTOR(content));

-- Example 2: Specify per-call parameters
INSERT INTO output_table
SELECT
  id,
  content,
  response AS sentiment
FROM ML_PREDICT(
  TABLE input_table,
  MODEL sentiment_model,
  DESCRIPTOR(content),
  MAP['user-prompt', 'Reply in Chinese']);

Résultat de l'exemple 1 :

Résultat de l'exemple 2 :

Image

L'exemple suivant enregistre un modèle multimodal et classe les images d'entrée.

La colonne d'entrée d'image (lorsque content-type est image_url) est de type STRING et prend en charge les formats suivants :

Si les images sont stockées dans des colonnes Blob d'un catalogue Paimon, vous pouvez utiliser la fonctionnalité d'URL signée temporaire (presigned URL) de Paimon Blob pour générer une URL publique en tant qu'entrée. Pour plus d'informations, consultez la page Apache Paimon.

-- Register a built-in model with content-type set to image_url
CREATE TEMPORARY MODEL sentiment_model
INPUT (prompt STRING)
OUTPUT (response STRING)
WITH (
  'provider' = 'openai-compat',
  'task' = 'chat/completions',
  'model' = 'qwen3.6-flash',
  'content-type' = 'image_url'
);

-- Use ML_PREDICT for real-time inference
INSERT INTO output_table
SELECT
  id,
  content,
  response AS sentiment
FROM ML_PREDICT(
  TABLE input_table,
  MODEL sentiment_model,
  DESCRIPTOR(content));

Vidéo

L'exemple suivant enregistre un modèle multimodal pour comprendre une vidéo d'entrée.

CREATE TEMPORARY MODEL sentiment_model
INPUT (video_url STRING)
OUTPUT (response STRING)
WITH (
  'provider' = 'openai-compat',
  'task' = 'chat/completions',
  'model' = 'qwen3.6-plus',
  'content-types' = 'video_url',
  'system-prompt' = 'Briefly describe the video content'
);

CREATE TEMPORARY VIEW input_table(content)
AS VALUES ('https://help-static-aliyun-doc.aliyuncs.com/file-manage-files/zh-CN/20251111/ckgcxt/%E5%8D%95%E4%BA%BA%E8%AF%B4%E8%AF%9D-1.mp4');

SELECT
  content,
  response AS sentiment
FROM ML_PREDICT(
  TABLE input_table,
  MODEL sentiment_model,
  DESCRIPTOR(content));

Texte + image unique

L'exemple suivant enregistre un modèle multimodal pour une inférence basée sur le chat avec des entrées textuelles et visuelles.

CREATE MODEL vl_model
INPUT (text_input STRING, image_input STRING)
OUTPUT (content STRING)
WITH (
  'provider' = 'openai-compat',
  'model' = 'qwen3.5-plus',
  'task' = 'chat/completions',
  'content-types' = 'text;image_url'
);

INSERT INTO result_sink
SELECT content FROM TABLE(ML_PREDICT(
  TABLE image_source,
  MODEL vl_model,
  DESCRIPTOR(text_input, image_input)
));         

Texte + plusieurs images (plusieurs colonnes)

Transmettez chaque image via une colonne INPUT distincte. Spécifiez image_url pour chaque colonne d'image dans content-types.

CREATE MODEL vl_model_multi
INPUT (prompt STRING, img1 STRING, img2 STRING)
OUTPUT (content STRING)
WITH (
  'provider' = 'openai-compat',
  'model' = 'qwen3.5-plus',
  'task' = 'chat/completions',
  'content-types' = 'text;image_url;image_url'
);

SELECT content FROM TABLE(ML_PREDICT(
  TABLE my_source,
  MODEL vl_model_multi,
  DESCRIPTOR(prompt, img1, img2)
));          

Texte + plusieurs images (tableau)

Transmettez plusieurs images dans une colonne ARRAY<STRING>. Utilisez multi_image_urls dans content-types.

CREATE MODEL vl_model_array
INPUT (prompt STRING, images ARRAY<STRING>)
OUTPUT (content STRING)
WITH (
  'provider' = 'openai-compat',
  'model' = 'qwen3.5-plus',
  'task' = 'chat/completions',
  'content-types' = 'text;multi_image_urls'
);

SELECT content FROM TABLE(ML_PREDICT(
  TABLE my_source,
  MODEL vl_model_array,
  DESCRIPTOR(prompt, images)
)); 

Texte + vidéo

CREATE TEMPORARY MODEL sentiment_model
INPUT (video_url STRING, question STRING)
OUTPUT (response STRING)
WITH (
  'provider' = 'openai-compat',
  'task' = 'chat/completions',
  'model' = 'qwen3.6-plus',
  'content-types' = 'video_url;text'
);

CREATE TEMPORARY VIEW input_table(content, question)
AS VALUES ('https://help-static-aliyun-doc.aliyuncs.com/file-manage-files/zh-CN/20251111/ckgcxt/%E5%8D%95%E4%BA%BA%E8%AF%B4%E8%AF%9D-1.mp4', 'How many people are there?');

SELECT
  content,
  response AS sentiment
FROM ML_PREDICT(
  TABLE input_table,
  MODEL sentiment_model,
  DESCRIPTOR(content, question));

-- When content-type / content-types are not specified in CREATE MODEL, the default is content-type = text
CREATE MODEL model_single
INPUT (input STRING)
OUTPUT (content STRING)
WITH (
  'provider' = 'openai-compat',
  'model' = 'qwen3.5-plus',
  'task' = 'chat/completions'
);

-- Call 1: Use MODEL parameters, content type is text
SELECT content FROM TABLE(ML_PREDICT(
  TABLE source_text,
  MODEL model_single,
  DESCRIPTOR(input)
));

-- Call 2: Override content type to image at call time
SELECT content FROM TABLE(ML_PREDICT(
  TABLE source_img,
  MODEL model_single,
  DESCRIPTOR(input),
  MAP['content-type', 'image_url']
));

-- Set default configuration in CREATE MODEL (multi-column model)
CREATE MODEL vl_model
INPUT (text_input STRING, image_input STRING)
OUTPUT (content STRING)
WITH (
  'provider' = 'openai-compat',
  'model' = 'qwen3.5-plus',
  'task' = 'chat/completions',
  'content-types' = 'text;image_url',
  'user-prompt' = 'Describe the image'
);

-- Call 1: Use default configuration
SELECT content FROM TABLE(ML_PREDICT(
  TABLE source_a, MODEL vl_model, DESCRIPTOR(text_input, image_input)
));

-- Call 2: Override user-prompt and column-level parameters
SELECT content FROM TABLE(ML_PREDICT(
  TABLE source_b, MODEL vl_model, DESCRIPTOR(text_input, image_input),
  MAP[
    'user-prompt', 'Answer in English',
    'image_input.min_pixels', '100',
    'image_input.max_pixels', '5000'
  ]
));

-- Call 3: Override content-types to change the type combination (treat both columns as text)
SELECT content FROM TABLE(ML_PREDICT(
  TABLE source_c, MODEL vl_model, DESCRIPTOR(text_input, image_input),
  MAP['content-types', 'text;text']
));

-- Call 4: Override content-types to change the type combination (reverse order)
SELECT content FROM TABLE(ML_PREDICT(
  TABLE source_c, MODEL vl_model, DESCRIPTOR(text_input, image_input),
  MAP['content-types', 'image_url;text']
));