Realtime Compute for Apache Flink:Invocación general

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

Parámetros de tipo de contenido

• Para entradas de una sola columna, use content-type para especificar el tipo de contenido. Valores compatibles: text, image_url.

• Para entradas de varias columnas, use content-types para especificar el tipo de contenido de cada columna. A continuación se muestran los valores compatibles y su asignación a los tipos de columna de Flink SQL:

• Se recomienda especificar content-types en los parámetros. Si establece tanto content-type como content-types, el framework reporta un error. Nota: las configuraciones de content-type y content-types en los parámetros del modelo y en la configuración por llamada se afectan mutuamente:

Parámetros a nivel de columna

Procesamiento por paquetes

• Procesamiento: el procesamiento por paquetes primero agrupa varios registros de entrada en un lote y luego llama al modelo de IA para ese lote. Esto reduce la cantidad de llamadas al modelo, pero aumenta el tiempo de inference por solicitud. Cuando los registros se pueden agrupar de forma eficaz, este método mejora significativamente el rendimiento del operador.

• Descripción de parámetros: bundle-size especifica la cantidad máxima de registros acumulados en un lote. Por su parte, bundle-allow-latency indica el tiempo de espera máximo cuando un lote no ha alcanzado el valor de bundle-size. El procesamiento comienza cuando se cumple la condición de tamaño de lote o la de tiempo de espera.

• Requisitos: el procesamiento por paquetes surte efecto según el esquema de entrada y salida del modelo.

  • Generación de texto: para los modelos cuya task es chat/completions, la entrada contiene exactamente una columna STRING de texto y la salida contiene exactamente una columna STRING o VARIANT. No se admiten entradas de varias columnas ni entradas multimodales como imágenes, videos o audio.

  • Embedding de texto: para los modelos cuya task es embeddings, la entrada contiene exactamente una columna STRING y la salida contiene exactamente una columna ARRAY<FLOAT>.

• Funciones compatibles: ML_PREDICT, text classification (AI_CLASSIFY), sentiment analysis (AI_SENTIMENT), information extraction (AI_EXTRACT), text summarization (AI_SUMMARIZE), text translation (AI_TRANSLATE), data masking (AI_MASK) y text embedding (AI_EMBED). De estas, ML_PREDICT admite el procesamiento por paquetes solo cuando el modelo registrado cumple los requisitos anteriores.

• Configuración global: esta configuración aplica a las llamadas de ML_PREDICT y a las llamadas de las funciones de IA específicas de dominio listadas en el job que cumplen los requisitos anteriores. Use los siguientes parámetros globales para habilitar el procesamiento por paquetes:

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

• Configuración por llamada: estos parámetros aplican únicamente a la llamada de función donde se especifican. Tanto ML_PREDICT como las funciones de IA específicas de dominio listadas usan el mismo método de configuración. Para consultar los parámetros y ejemplos, consulte Per-call configuration.

Límites

• Solo es compatible con VVR 11.9 y versiones posteriores.

• No es posible habilitar el manejo de errores cuando las llamadas al modelo se ejecutan en modo síncrono.

• Por ahora, el manejo de errores no se puede usar junto con Vectorization result cache (CACHE_TABLE).

Habilitar el manejo de errores

Agregue el parámetro FAIL_ON_ERROR => FALSE cuando llame a una función de IA.

Requisito previo: conserve la configuración de reintentos predeterminada del modelo: error-handling-strategy='retry' y retry-fallback-strategy='failover'. Para obtener más información sobre los parámetros del modelo, consulte Model settings.

Cambios en la salida

Después de habilitar el manejo de errores, el sistema agrega una columna de error err_msg de tipo VARIANT anulable al final de los campos de salida originales. Esta columna la genera el sistema, por lo que no necesita declararla en la tabla de entrada ni en la cláusula OUTPUT de CREATE MODEL.

• Llamada exitosa: err_msg es NULL.

• Llamada fallida definitiva: todos los campos de resultado del modelo son NULL y err_msg contiene la información del error.

Si usa SELECT *, el resultado incluye la nueva columna de error. Si selecciona los campos de forma explícita, debe especificar la columna manualmente. El nombre predeterminado de la columna es err_msg. Si el nombre entra en conflicto con una columna de salida existente, el sistema agrega un sufijo numérico, como err_msg0. Use el nombre real de la columna para hacer referencia a ella.

Ejemplos

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)

Cómo se maneja cada tipo de error

El siguiente comportamiento aplica cuando el manejo de errores está habilitado y se conserva la configuración de reintentos predeterminada. Una fila de error es aquella que conserva la entrada, devuelve resultados de modelo vacíos y registra el motivo del fallo en err_msg.

Texto

En el siguiente ejemplo se registra y se usa un modelo integrado de Flink AI Service para clasificar el sentimiento de un texto de entrada. El ejemplo 1 usa parámetros a nivel de MODEL, mientras que el ejemplo 2 sobrescribe user-prompt en el momento de la llamada.

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

Salida del ejemplo 1:

Salida del ejemplo 2:

Imagen

En el siguiente ejemplo se registra un modelo multimodal y se clasifican imágenes de entrada.

La columna de entrada de imagen (cuando content-type es image_url) es de tipo STRING y admite los siguientes formatos:

Si las imágenes se almacenan en columnas Blob de un Paimon Catalog, puede usar la función de URL prefirmada de Paimon Blob para generar una URL pública como entrada. Para obtener más información, consulte 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));

Video

En el siguiente ejemplo se registra un modelo multimodal para comprender un video de entrada.

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

Texto + una sola imagen

En el siguiente ejemplo se registra un modelo multimodal para realizar inference basada en chat con entradas de texto e imagen.

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

Texto + varias imágenes (varias columnas)

Pase cada imagen mediante una columna INPUT independiente. Especifique image_url para cada columna de imagen en 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)
));          

Texto + varias imágenes (arreglo)

Pase varias imágenes en una columna ARRAY<STRING>. Use multi_image_urls en 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)
)); 

Texto + video

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