AI_EXTRACT uses Flink operators with Large Language Model (LLM) services to extract structured information. It requires Ververica Runtime (VVR) 11.4 or later and is subject to rate limiting by the model platform.-Realtime Compute for Apache Flink(Flink)-阿里云帮助中心

Limitations

Supported only by the real-time computing engine Ververica Runtime (VVR) 11.4 or later.
To use the Flink AI service (built-in models), VVR 11.7 or later is required, and the Flink AI service must be activated. For details, see Flink AI service (built-in models).
Throughput is limited by the model service platform. When traffic reaches the platform's access limit, backpressure occurs on the Flink operators used in AI_EXTRACT statements. Severe rate limiting can trigger timeouts and cause the Flink job to restart.

Syntax

AI_EXTRACT(
  MODEL => MODEL <MODEL NAME>,
  INPUT => <INPUT COLUMN NAME>,
  EXTRACT_SCHEMA => <EXTRACT SCHEMA>
)

Parameters

Parameter	Data type	Description
`MODEL <MODEL NAME>`	MODEL	The name of the registered model service. For more information, see Model Settings. Note: The output type of this model must be `VARIANT`.
`<INPUT COLUMN NAME>`	STRING	The column containing the raw text to extract information from.
`<EXTRACT_SCHEMA>`	STRING	A JSON string that defines the fields to extract and their data types. Must be a constant.

Return values

Column	Data type	Description
`extracted_json`	STRING	The extracted fields as a JSON string.

Examples

Test data

id	description
1	Xiao Ming is 18 years old and lives in Hangzhou. His phone number is ******.

SQL statements

The following example references a Flink built-in model and uses AI_EXTRACT to extract user information.

CREATE TEMPORARY MODEL general_model
INPUT (`input` STRING)
OUTPUT (`content` VARIANT)
WITH (
    'provider' = 'openai-compat',
    'task'     = 'chat/completions',
    'model'    = 'qwen3.6-flash'
);

CREATE TEMPORARY VIEW infos(id, description)
AS VALUES (1, 'Xiao Ming is 18 years old and lives in Hangzhou. His phone number is ******.');

-- Positional argument style
SELECT id, extracted_json
FROM infos,
LATERAL TABLE(
  AI_EXTRACT(
    MODEL general_model,
    description,
    '{"name":"string","phone":"string","address":"string","age":"int"}'));

-- Named argument style
SELECT id, extracted_json
FROM infos,
LATERAL TABLE(
  AI_EXTRACT(
    MODEL => MODEL general_model,
    INPUT => description,
    EXTRACT_SCHEMA => '{"name":"string","phone":"string","address":"string","age":"int"}'));

Output

id	extracted_json
1	`{"address":"Hangzhou","age":18,"name":"Xiao Ming","phone":"******"}`