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 à colonne unique, 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 répertorié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)

• 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.

• Type de données : Seul le type text est pris en charge.

• Fonctions prises en charge : ML_PREDICT, classification de texte (AI_CLASSIFY), analyse de sentiment (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).

• Configuration globale : Cette configuration s'applique à chaque appel ML_PREDICT et à chaque appel aux fonctions d'IA spécifiques listées dans la tâche. Utilisez les paramètres globaux suivants pour activer le traitement par lots :

  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 listées utilisent la même méthode de configuration. Pour les paramètres et les exemples, consultez Configuration par appel.

Texte

L'exemple suivant enregistre et utilise un modèle intégré à Flink AI Service pour classer le texte d'entrée selon son sentiment. 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.

-- 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 analyser 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']
));