Quick start: Real-time data analysis with LLMs - Realtime Compute for Apache Flink

Background information

Model Studio is a platform for building LLM-powered applications. It integrates with Realtime Compute for Apache Flink, enabling you to combine LLM capabilities with Flink real-time data pipelines. Two core model types are supported:

chat/completions model An LLM for dialogue generation and text understanding, used in sentiment analysis, intent recognition, and question-answering systems.
- Sentiment analysis: Classifies social media comments in real time as positive, negative, or neutral.
- Intelligent customer service: Powers natural language interactions for automated customer service systems.
- Content moderation: Detects sensitive content or policy violations in text for content security audits.
embedding model Converts text into high-dimensional vector representations for semantic search, recommendation systems, and knowledge graph construction.
- Semantic search: Vectorizes product descriptions or user queries for relevance-based search.
- Recommendation systems: Discovers associations between user interests and product features through text vectorization.
- Knowledge graph: Converts unstructured text into vectors for knowledge extraction and relationship modeling.

Prerequisites

A Flink workspace is activated. Activate Realtime Compute for Apache Flink.
A Model Studio workspace is activated with network access to the Flink development console. Access the API of a model or application in Model Studio over a private network.
If you access Model Studio by domain name, register it in the Flink development console. Manage domain names.

Limitations

This feature is supported only in Ververica Runtime (VVR) 11.1 and later.

Step 1: Register a Model Studio model

Register a model by following Configure a model.

Chat/completions model

The following sample SQL registers the qwen-turbo model:

CREATE MODEL ai_analyze_sentiment
INPUT (`input` STRING)
OUTPUT (`content` STRING)
WITH (
    'provider'='bailian',
    'endpoint'='<base_url>/compatible-mode/v1/chat/completions',    -- The endpoint for chat/completions model tasks.
    'api-key' = '<YOUR KEY>',
    'model'='qwen-turbo',                                                               -- The Qwen-turbo model.
    'system-prompt' = 'Classify the text below into one of the following labels: [positive, negative, neutral, mixed]. Output only the label.'
);

Replace <base_url> in the endpoint value based on your access method:

Internet access: replace <base_url> with https://dashscope-intl.aliyuncs.com.
VPC private network access: replace <base_url> with https://vpc-ap-southeast-1.dashscope.aliyuncs.com.

Note
The endpoint parameter supports only the HTTPS protocol.

Embedding model

The following sample SQL registers the text-embedding-v3 model:

CREATE MODEL embedding_model
INPUT (`input` STRING)
OUTPUT (`embeddings` ARRAY<FLOAT>)
WITH (
    'provider'='bailian',
    'endpoint'='https://dashscope.aliyuncs.com/compatible-mode/v1/embeddings',   -- The endpoint for embedding model tasks.
    'api-key' = '<YOUR KEY>',
    'model'='text-embedding-v3'                                                  -- The text-embedding-v3 model.
);

Step 2: Create a job

Create a draft SQL streaming job. Flink SQL jobs.

Step 3: Write an SQL job for LLM analysis

Chat/completions model

Call the registered ai_analyze_sentiment model with ML_PREDICT to perform sentiment analysis on movie reviews.

Important

The ML_PREDICT operator is subject to Model Studio rate limits. When limits are reached, backpressure occurs and the ML_PREDICT operator becomes the bottleneck. Severe throttling can cause operator timeouts and job restarts. Check model-specific rate limits in Limits on QPS and tokens. Contact your business manager to request limit increases.

Copy this SQL to the editor.

-- Create a temporary sink table.
CREATE TEMPORARY TABLE print_sink(
  id BIGINT,
  movie_name VARCHAR, 
  predict_label VARCHAR, 
  actual_label VARCHAR
) WITH (
  'connector' = 'print',   -- Use the print connector.
  'logger' = 'true'        -- Display the results in the console.
);
-- Create a temporary data view to construct test data.
-- | id | movie_name         | comment                                                                                                                        | actual_label |
-- | 1  | Her Story          | My favorite part was when the kid guessed the sounds. It is one of the most romantic narratives I have seen in movies. Very gentle and loving. | POSITIVE     |
-- | 2  | The Dumpling Queen | Unremarkable.                                                                                                                  | NEGATIVE     |
CREATE TEMPORARY VIEW movie_comment(id, movie_name, user_comment, actual_label)
AS VALUES (1, 'Her Story', 'My favorite part was when the kid guessed the sounds. It is one of the most romantic narratives I have seen in movies. Very gentle and loving.', 'positive'), (2, 'The Dumpling Queen', 'Unremarkable.', 'negative');
INSERT INTO print_sink
SELECT id, movie_name, content as predict_label, actual_label 
FROM ML_PREDICT(
  TABLE movie_comment, 
  MODEL ai_analyze_sentiment,  -- The registered Qwen qwen-turbo model.
  DESCRIPTOR(user_comment));

Embedding model

Call the registered embedding_model with ML_PREDICT to generate embeddings for movie reviews and write results to Milvus (public preview).

Important

The ML_PREDICT operator is subject to Model Studio rate limits. When limits are reached, backpressure occurs and the ML_PREDICT operator becomes the bottleneck. Severe throttling can cause operator timeouts and job restarts. Check model-specific rate limits in Limits on QPS and tokens. Contact your business manager to request limit increases.

Copy this SQL to the editor.

-- Create a temporary sink table named milvus_sink.
CREATE TEMPORARY TABLE milvus_sink
(
    id STRING,
    movie_name STRING,
    user_comment STRING,
    embeddings ARRAY<FLOAT>,
    PRIMARY KEY (id) NOT ENFORCED
)
WITH (
    'connector' = 'milvus',
    'endpoint' = '<YOUR-ENDPOINT>',
    'port' = '<YOUR-PORT>',
    'userName' = '<YOUR-USERNAME>',
    'password' = '<YOUR-PASSWORD>',
    'databaseName' = 'default',
    'collectionName' = 'movie-comment-embeddings'
);
-- Create a temporary data view to construct test data.
-- | id | movie_name         | comment                                                                                                                        |
-- | 1  | Her Story          | My favorite part was when the kid guessed the sounds. It is one of the most romantic narratives I have seen in movies. Very gentle and loving. |
-- | 2  | The Dumpling Queen | Unremarkable.                                                                                                                  |
CREATE TEMPORARY VIEW movie_comment(id, movie_name,  user_comment)
AS VALUES ('1', 'Her Story', 'My favorite part was when the kid guessed the sounds. It is one of the most romantic narratives I have seen in movies. Very gentle and loving.'), ('2', 'The Dumpling Queen', 'Unremarkable.');
INSERT INTO
    milvus_sink
SELECT
    id,
    movie_name,
    user_comment,
    embeddings
FROM
    ML_PREDICT (
        TABLE movie_comment,
        MODEL embedding_model,  -- The registered text-embedding-v3 model.                   
        DESCRIPTOR (user_comment)
    );

Step 4: Deploy and start the job

Deploy and start the job. Flink SQL jobs.

Step 5: View the analysis results

Chat/completions model

Verify that the job status is FINISHED.
In the O&M console, go to Deployments and click the target job.
On the Logs tab, click the Task Managers subtab and select the current TaskManager.
Click Log and search for PrintSinkOutputWriter.

The predict_label should match the actual_label.

For example, +I[1, Her Story, positive, positive] and +I[2, The Dumpling Queen, negative, negative] confirm that predictions match actual labels.

Realtime Compute for Apache Flink:Quick start: Real-time data analysis with LLMs

Background information

Prerequisites

Limitations

Step 1: Register a Model Studio model

Chat/completions model

Embedding model

Step 2: Create a job

Step 3: Write an SQL job for LLM analysis

Chat/completions model

Embedding model

Step 4: Deploy and start the job

Step 5: View the analysis results

Chat/completions model

Related documents