Call AI functions in PyFlink

Prerequisites

• You have activated Realtime Compute for Apache Flink and created a workspace. For more information, see Activate Realtime Compute for Apache Flink. The Realtime Compute engine version must be VVR 11.7 or later.

• You have activated the Flink AI service. For more information, see Flink AI service (built-in models).

• (Optional) Configure a local development environment. Install the VVR PyFlink dependency package by running the following command:

  pip install ververica-flink

  Note

  ververica-flink is an API-Only package that provides only type definitions and interface declarations for local development. Jobs must still be submitted on the Realtime Compute for Apache Flink platform.

Supported functions

AI Functions are built-in Flink Table API functions that let you call a large language model directly within a stream or batch processing pipeline. For more information about supported functions, see AI large language model integration.

This topic uses ai_classify as an example to demonstrate the complete process. The invocation pattern is the same for other functions, but the parameters are different.

Step 1: Create a Python script and initialize environment

Create a Python file, such as ai_classify_example.py, to import the required modules and initialize the Flink execution environment.

from pyflink.datastream import StreamExecutionEnvironment
from pyflink.table import StreamTableEnvironment, DataTypes, Schema, ModelDescriptor
from pyflink.table.ai_functions import ai_classify
from pyflink.table.expressions import col

env = StreamExecutionEnvironment.get_execution_environment()
t_env = StreamTableEnvironment.create(env)

The StreamTableEnvironment is the entry point for the Table API, and all subsequent model registration, data operations, and job execution are performed through this object.

Step 2: Register a model

Use ModelDescriptor to define the LLM connection information, and use create_temporary_model to register it in the Table environment.

model_descriptor = (
    ModelDescriptor.for_provider("openai-compat")
    .input_schema(
        Schema.new_builder()
        .column("input", DataTypes.STRING())
        .build()
    )
    .output_schema(
        Schema.new_builder()
        .column("content", DataTypes.VARIANT())
        .build()
    )
    .option("task", "chat/completions")
    .option("model", "qwen3.6-flash")
    .option("temperature", "0.0")
    .build()
)

t_env.create_temporary_model("classify_model", model_descriptor, True)
model = t_env.from_model("classify_model")

Parameters

When the third parameter of create_temporary_model is True, an existing model with the same name is overwritten. from_model returns a model reference for subsequent AI Function calls. For more information about the parameters, see Model settings.

Register a persistent model

Models registered by using create_temporary_model are valid only for the current session and are destroyed after the job ends. To persist a model to a catalog for reuse across multiple jobs, use create_model and specify the full three-part name:

t_env.create_model("my_catalog.my_db.classify_model", model_descriptor, True)

When referencing a registered persistent model, also use its three-part name:

model = t_env.from_model("my_catalog.my_db.classify_model")

For a persistent model, the connection information and schema definition are stored in the catalog, so you do not need to declare them in every job.

Step 3: Prepare the input data

Define the data schema and create an input table.

input_schema = DataTypes.ROW([
    DataTypes.FIELD("id", DataTypes.BIGINT()),
    DataTypes.FIELD("review", DataTypes.STRING()),
])

reviews = [
    (1, "This product is absolutely amazing, I love it!"),
    (2, "Terrible quality, broke after one day of use."),
    (3, "It's okay, nothing special but gets the job done."),
    (4, "Fast delivery, great packaging, very satisfied!"),
    (5, "Would not recommend, customer service was unhelpful."),
]

input_table = t_env.from_elements(reviews, input_schema)

from_elements converts a Python list into a Flink table. In a production environment, input data usually comes from connectors such as Kafka and databases.

Step 4: Call the AI function

Flink provides a rich set of built-in AI Functions.

For example, AI_classify returns a UDTF (user-defined table function) expression, which is expanded and appended after the original table columns by using join_lateral.

labels = ["positive", "negative", "neutral"]

classify_expr = ai_classify(model, col("review"), labels)

result_table = input_table.join_lateral(classify_expr)

ai_classify takes three parameters:

After join_lateral is executed, the column structure of the result table is as follows:

Step 5: Execute and view results

Call execute().collect() to trigger job execution and collect the results.

results = list(result_table.execute().collect())
results.sort(key=lambda r: r[0])

for row in results:
    review_id, review_text, category, confidence = row
    print(f"id={review_id} category={category:<12s} confidence={confidence:.3f} review={review_text}")

Step 6: Deploy the job

1. Deploy and start the Python job. For deployment instructions, see Flink Python jobs.

2. In the Flink job settings, navigate to Configuration > Parameters > Other Configuration and add the following configuration:

   pipeline.used-builtin-models: openai-compat

3. View the output on the Job Logs page. Go to the target job's Task Managers > Logs page and search for the output.

   Example output:

   id=1 category=positive     confidence=0.950 review=This product is absolutely amazing, I love it!
   id=2 category=negative     confidence=0.980 review=Terrible quality, broke after one day of use.
   id=3 category=neutral      confidence=0.850 review=It's okay, nothing special but gets the job done.
   id=4 category=positive     confidence=0.960 review=Fast delivery, great packaging, very satisfied!
   id=5 category=negative     confidence=0.920 review=Would not recommend, customer service was unhelpful.

Complete code

import logging

from pyflink.datastream import StreamExecutionEnvironment
from pyflink.table import StreamTableEnvironment, DataTypes, Schema, ModelDescriptor
from pyflink.table.ai_functions import ai_classify
from pyflink.table.expressions import col

logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s")
logger = logging.getLogger(__name__)


def main():
    env = StreamExecutionEnvironment.get_execution_environment()
    t_env = StreamTableEnvironment.create(env)

    model_descriptor = (
        ModelDescriptor.for_provider("openai-compat")
        .input_schema(
            Schema.new_builder()
            .column("input", DataTypes.STRING())
            .build()
        )
        .output_schema(
            Schema.new_builder()
            .column("content", DataTypes.VARIANT())
            .build()
        )
        .option("task", "chat/completions")
        .option("model", "qwen3.6-flash")
        .option("temperature", "0.0")
        .build()
    )

    t_env.create_temporary_model("classify_model", model_descriptor, True)
    model = t_env.from_model("classify_model")

    input_schema = DataTypes.ROW([
        DataTypes.FIELD("id", DataTypes.BIGINT()),
        DataTypes.FIELD("review", DataTypes.STRING()),
    ])

    reviews = [
        (1, "This product is absolutely amazing, I love it!"),
        (2, "Terrible quality, broke after one day of use."),
        (3, "It's okay, nothing special but gets the job done."),
        (4, "Fast delivery, great packaging, very satisfied!"),
        (5, "Would not recommend, customer service was unhelpful."),
    ]

    input_table = t_env.from_elements(reviews, input_schema)

    labels = ["positive", "negative", "neutral"]
    classify_expr = ai_classify(model, col("review"), labels)
    result_table = input_table.join_lateral(classify_expr)

    results = list(result_table.execute().collect())
    results.sort(key=lambda r: r[0])

    for row in results:
        review_id, review_text, category, confidence = row
        logger.info("id=%d category=%-12s confidence=%.3f review=%s",
                    review_id, category, confidence, review_text)


if __name__ == "__main__":
    main()