Build Real-Time AI Inference Pipelines with ML_PREDICT in Flink SQL - Realtime Compute for Apache Flink

ML_PREDICT runs AI model inference inside a Flink SQL query, passing rows from a table or view to a registered model and returning the model's output as additional columns.

Limitations

Requires Ververica Runtime (VVR) 11.1 or later.
Throughput is subject to the rate limits of Alibaba Cloud Model Studio. When the rate limits are reached, the Flink job experiences back pressure with ML_PREDICT operators as the bottleneck. In severe cases, timeout errors and job restarts may occur.

Syntax

ML_PREDICT(TABLE <TABLE NAME>, MODEL <MODEL NAME>, DESCRIPTOR(<INPUT COLUMN NAMES>))

Parameters

Parameter	Type	Description
`TABLE <TABLE NAME>`	TABLE	The input stream for model inference. Can be a physical table or a view.
`MODEL <MODEL NAME>`	MODEL	The name of the registered model.
`DESCRIPTOR(<INPUT COLUMN NAMES>)`	N/A	The columns in the input data to pass to the model.

Examples

This example registers the Qwen-Turbo model and uses ML_PREDICT to classify movie review sentiment.

Test data

id	movie_name	user_comment	actual_label
1	Her Story	My favourite part was when the kid guessed the sounds. It was romantic, heartwarming, and loving.	POSITIVE
2	The Dumpling Queen	Nothing special.	NEGATIVE

SQL statements

The following statements register the model, create a test view, and run inference:

-- Register the Qwen-Turbo model with a sentiment classification prompt
CREATE TEMPORARY MODEL ai_analyze_sentiment
INPUT (`input` STRING)
OUTPUT (`content` STRING)   -- The output column name 'content' is defined here
WITH (
    'provider'='bailian',
    'endpoint'='<YOUR ENDPOINT>',
    'apiKey' = '<YOUR KEY>',
    'model'='qwen-turbo',
    'systemPrompt' = 'Classify the text below into one of the following labels: [positive, negative, neutral, mixed]. Output only the label.'
);

-- Create a view with test data
CREATE TEMPORARY VIEW movie_comment(id, movie_name, user_comment, actual_label)
AS VALUES
    (1, 'Her Story', 'My favourite part was when the kid guessed the sounds. It was romantic, heartwarming, and loving.', 'positive'),
    (2, 'The Dumpling Queen', 'Nothing special.', 'negative');

-- Run inference; 'content' in the SELECT comes from the OUTPUT definition above
SELECT id, movie_name, content AS predict_label, actual_label
FROM ML_PREDICT(TABLE movie_comment, MODEL ai_analyze_sentiment, DESCRIPTOR(user_comment));

Output

The predict_label column matches the actual_label column, confirming that the model correctly classified both reviews.

id	movie_name	predict_label	actual_label
1	Her Story	POSITIVE	POSITIVE
2	The Dumpling Queen	NEGATIVE	NEGATIVE